From bf837385044541d47441d9e70643e9902f8bef25 Mon Sep 17 00:00:00 2001 From: Gabriel Mechali Date: Wed, 8 Jul 2026 17:55:15 -0400 Subject: [PATCH 01/10] Move the preprocessing jobi n the workflow. --- infra/dcp/main.tf | 6 +- .../modules/datacommons_services/variables.tf | 2 +- .../modules/ingestion/workflow/workflow.yaml | 78 +++++++++++ infra/dcp/modules/stack/main.tf | 127 +++++++++--------- 4 files changed, 145 insertions(+), 68 deletions(-) diff --git a/infra/dcp/main.tf b/infra/dcp/main.tf index 4af80ac9..c7cb6691 100644 --- a/infra/dcp/main.tf +++ b/infra/dcp/main.tf @@ -138,9 +138,9 @@ locals { helper_service_image = var.ingestion_helper_service_image # Dataflow Network Configuration - dataflow_ip_configuration = var.ingestion_dataflow_ip_configuration - dataflow_subnetwork = var.ingestion_dataflow_subnetwork - dataflow_template_gcs_path = var.ingestion_dataflow_template_gcs_path + dataflow_ip_configuration = var.ingestion_dataflow_ip_configuration + dataflow_subnetwork = var.ingestion_dataflow_subnetwork + dataflow_template_gcs_path = var.ingestion_dataflow_template_gcs_path } } diff --git a/infra/dcp/modules/datacommons_services/variables.tf b/infra/dcp/modules/datacommons_services/variables.tf index 8d3c748a..c3d73740 100644 --- a/infra/dcp/modules/datacommons_services/variables.tf +++ b/infra/dcp/modules/datacommons_services/variables.tf @@ -71,7 +71,7 @@ variable "mcp_instructions_path" { } variable "resolve_with_spanner_embeddings" { - type = bool + type = bool } # ============================================================================= diff --git a/infra/dcp/modules/ingestion/workflow/workflow.yaml b/infra/dcp/modules/ingestion/workflow/workflow.yaml index fd757dcc..55c9c2ab 100644 --- a/infra/dcp/modules/ingestion/workflow/workflow.yaml +++ b/infra/dcp/modules/ingestion/workflow/workflow.yaml @@ -20,6 +20,8 @@ main: - decoded_imports: '$${json.decode(input.importList)}' - num_imports: '$${len(decoded_imports)}' - combined_import_name: "" + - prep_job_name: '${clean_namespace_prefix}dc-ingestion-preprocessing-job' + - comma_separated_imports: "" - imports_status_list: [] - imports_history_list: [] - imports_version_list: [] @@ -48,6 +50,7 @@ main: - clean_item: '$${text.replace_all(text.replace_all(text.to_lower(imp.importName), "/", "-"), "_", "-")}' - join_prefix: '$${if(combined_import_name == "", "", "-")}' - combined_import_name: '$${combined_import_name + join_prefix + clean_item}' + - comma_separated_imports: '$${if(comma_separated_imports == "", imp.importName, comma_separated_imports + "," + imp.importName)}' - set_dataflow_job_name: assign: - import_name_len: '$${len(combined_import_name)}' @@ -108,6 +111,29 @@ main: imports: '$${imports_status_list}' status: 'STAGING' + - run_preprocessing: + call: launch_and_poll_preprocessing_job + args: + job_name: '$${prep_job_name}' + workflow_id: '$${workflow_id}' + temp_location: '$${input.tempLocation}' + imports_arg: '$${comma_separated_imports}' + region: '$${input.region}' + project_id: '${project_id}' + result: prep_res + + - read_handshake_file: + call: googleapis.storage.v1.objects.get + args: + bucket: '$${bucket_name}' + object: '$${"datacommons/ingestion_records/" + workflow_id + ".json"}' + alt: 'media' + result: handshake_data + + - update_launch_params: + assign: + - launch_params.importList: '$${handshake_data.importList}' + - launch_dataflow: call: launch_and_poll_dataflow args: @@ -396,3 +422,55 @@ launch_and_poll_dataflow: return: job_id: $${job_id} launch_result: $${launch_result} + +launch_and_poll_preprocessing_job: + params: [job_name, workflow_id, temp_location, imports_arg, region, project_id] + steps: + - run_cloud_run_job: + call: googleapis.run.v2.projects.locations.jobs.run + args: + name: '$${"projects/" + project_id + "/locations/" + region + "/jobs/" + job_name}' + body: + overrides: + containerOverrides: + - env: + - name: 'WORKFLOW_EXECUTION_ID' + value: '$${workflow_id}' + - name: 'TEMP_LOCATION' + value: '$${temp_location}' + - name: 'DATA_RUN_MODE' + value: 'dcpbridge' + args: ['$${"--imports=" + imports_arg}'] + result: run_res + + - assign_execution_name: + assign: + - execution_name: '$${run_res.metadata.name}' + + - poll_execution: + steps: + - get_execution: + call: googleapis.run.v2.projects.locations.jobs.executions.get + args: + name: '$${execution_name}' + result: exec_status + - check_exec_terminal: + switch: + - condition: '$${default(map.get(exec_status, "succeededCount"), 0) > 0 or default(map.get(exec_status, "failedCount"), 0) > 0 or default(map.get(exec_status, "cancelledCount"), 0) > 0}' + next: check_exec_success + - wait_for_execution: + call: sys.sleep + args: + seconds: 30 + next: poll_execution + + - check_exec_success: + switch: + - condition: '$${default(map.get(exec_status, "succeededCount"), 0) > 0}' + next: return_success + + - fail_on_exec_status: + raise: '$${"Preprocessing Cloud Run job failed or was cancelled: " + exec_status.name}' + + - return_success: + return: '$${exec_status}' diff --git a/infra/dcp/modules/stack/main.tf b/infra/dcp/modules/stack/main.tf index 13b9db04..9dcac086 100644 --- a/infra/dcp/modules/stack/main.tf +++ b/infra/dcp/modules/stack/main.tf @@ -105,10 +105,10 @@ module "storage" { source = "../storage" # Ingestion Workflow Bucket Vars - create_artifacts_bucket = var.storage_create_artifacts_bucket - artifacts_bucket_name = var.storage_artifacts_bucket_name - region = var.global.region - stateful_deletion_protection = var.global.stateful_deletion_protection + create_artifacts_bucket = var.storage_create_artifacts_bucket + artifacts_bucket_name = var.storage_artifacts_bucket_name + region = var.global.region + stateful_deletion_protection = var.global.stateful_deletion_protection # Shared vars project_id = var.global.project_id @@ -119,25 +119,25 @@ module "ingestion_preprocessing_job" { source = "../ingestion/preprocessing_job" count = var.ingestion_config.enable_ingestion ? 1 : 0 - project_id = var.global.project_id - namespace = var.global.namespace - region = var.global.region - stateless_deletion_protection = var.global.stateless_deletion_protection - image = var.ingestion_config.preprocessing_job_image - cpu = var.ingestion_config.preprocessing_job_cpu - memory = var.ingestion_config.preprocessing_job_memory - timeout = var.ingestion_config.preprocessing_job_timeout - vpc_connector_id = var.redis_config.enable ? module.redis[0].connector_id : null - bucket_name = module.storage.artifacts_bucket_name - input_path = var.ingestion_config.input_path - ingestion_artifacts_path = var.ingestion_config.ingestion_artifacts_path - run_database_init = false - use_spanner = true - enable_spanner_embeddings = var.datacommons_services_config.resolve_with_spanner_embeddings - env_vars = local.cloud_run_shared_env_variables - secret_env_vars = local.datacommons_services_secrets - dc_api_key_secret_id = module.auth.dc_api_key_secret_id - maps_api_key_secret_id = module.auth.maps_api_key_secret_id + project_id = var.global.project_id + namespace = var.global.namespace + region = var.global.region + stateless_deletion_protection = var.global.stateless_deletion_protection + image = var.ingestion_config.preprocessing_job_image + cpu = var.ingestion_config.preprocessing_job_cpu + memory = var.ingestion_config.preprocessing_job_memory + timeout = var.ingestion_config.preprocessing_job_timeout + vpc_connector_id = var.redis_config.enable ? module.redis[0].connector_id : null + bucket_name = module.storage.artifacts_bucket_name + input_path = var.ingestion_config.input_path + ingestion_artifacts_path = var.ingestion_config.ingestion_artifacts_path + run_database_init = false + use_spanner = true + enable_spanner_embeddings = var.datacommons_services_config.resolve_with_spanner_embeddings + env_vars = local.cloud_run_shared_env_variables + secret_env_vars = local.datacommons_services_secrets + dc_api_key_secret_id = module.auth.dc_api_key_secret_id + maps_api_key_secret_id = module.auth.maps_api_key_secret_id depends_on = [module.auth] } @@ -155,26 +155,26 @@ module "ingestion_dataflow" { module "ingestion_helper_service" { source = "../ingestion/helper_service" - deploy = var.ingestion_config.enable_ingestion - project_id = var.global.project_id - namespace = var.global.namespace - region = var.global.region + deploy = var.ingestion_config.enable_ingestion + project_id = var.global.project_id + namespace = var.global.namespace + region = var.global.region stateless_deletion_protection = var.global.stateless_deletion_protection # Use index [0] because module.spanner is conditional. Fallback to empty string if disabled. - spanner_instance_id = var.spanner_config.enable ? module.spanner[0].spanner_instance_id : "" - spanner_database_id = var.spanner_config.enable ? module.spanner[0].spanner_database_id : "" - bigquery_connection_id = var.spanner_config.enable ? module.spanner[0].bigquery_connection_id : "" - ingestion_bucket_name = module.storage.artifacts_bucket_name - image = var.ingestion_config.helper_service_image - use_spanner = var.spanner_config.enable + spanner_instance_id = var.spanner_config.enable ? module.spanner[0].spanner_instance_id : "" + spanner_database_id = var.spanner_config.enable ? module.spanner[0].spanner_database_id : "" + bigquery_connection_id = var.spanner_config.enable ? module.spanner[0].bigquery_connection_id : "" + ingestion_bucket_name = module.storage.artifacts_bucket_name + image = var.ingestion_config.helper_service_image + use_spanner = var.spanner_config.enable enable_bigquery_postprocessing = var.ingestion_config.workflow_enable_bigquery_postprocessing enable_bigquery_connection = var.spanner_config.enable_bigquery_connection enable_embeddings_generation = var.spanner_config.enable_embeddings_generation # Redis configuration for cache clearing - vpc_connector_id = var.redis_config.enable && length(module.redis) > 0 ? module.redis[0].connector_id : null - redis_host = var.redis_config.enable && length(module.redis) > 0 ? module.redis[0].redis_host : "" - redis_port = var.redis_config.enable && length(module.redis) > 0 ? tostring(module.redis[0].redis_port) : "" + vpc_connector_id = var.redis_config.enable && length(module.redis) > 0 ? module.redis[0].connector_id : null + redis_host = var.redis_config.enable && length(module.redis) > 0 ? module.redis[0].redis_host : "" + redis_port = var.redis_config.enable && length(module.redis) > 0 ? tostring(module.redis[0].redis_port) : "" ingestion_artifacts_path = "${var.ingestion_config.ingestion_artifacts_path}/metadata" } @@ -185,7 +185,7 @@ module "ingestion_workflow" { deploy = var.ingestion_config.enable_ingestion namespace = var.global.namespace region = var.global.region - stateless_deletion_protection = var.global.stateless_deletion_protection + stateless_deletion_protection = var.global.stateless_deletion_protection project_id = var.global.project_id lock_acquisition_timeout = var.ingestion_config.workflow_lock_acquisition_timeout ingestion_helper_url = module.ingestion_helper_service.ingestion_helper_url @@ -233,25 +233,25 @@ module "datacommons_services" { source = "../datacommons_services" count = var.datacommons_services_config.enable ? 1 : 0 - project_id = var.global.project_id - namespace = var.global.namespace - region = var.global.region - stateless_deletion_protection = var.global.stateless_deletion_protection - image = var.datacommons_services_config.image - cpu = var.datacommons_services_config.cpu - memory = var.datacommons_services_config.memory - min_instances = var.datacommons_services_config.min_instances - max_instances = var.datacommons_services_config.max_instances - make_public = var.datacommons_services_config.allow_unauthenticated_access - google_analytics_tag_id = var.datacommons_services_config.google_analytics_tag - mcp_search_scope = var.datacommons_services_config.search_scope - enable_mcp = var.datacommons_services_config.enable_mcp - mcp_instructions_path = var.datacommons_services_config.instructions_path - artifacts_bucket_name = module.storage.artifacts_bucket_name - vpc_connector_id = var.redis_config.enable ? module.redis[0].connector_id : null - use_spanner = var.spanner_config.enable - env_vars = local.cloud_run_shared_env_variables - secret_env_vars = local.datacommons_services_secrets + project_id = var.global.project_id + namespace = var.global.namespace + region = var.global.region + stateless_deletion_protection = var.global.stateless_deletion_protection + image = var.datacommons_services_config.image + cpu = var.datacommons_services_config.cpu + memory = var.datacommons_services_config.memory + min_instances = var.datacommons_services_config.min_instances + max_instances = var.datacommons_services_config.max_instances + make_public = var.datacommons_services_config.allow_unauthenticated_access + google_analytics_tag_id = var.datacommons_services_config.google_analytics_tag + mcp_search_scope = var.datacommons_services_config.search_scope + enable_mcp = var.datacommons_services_config.enable_mcp + mcp_instructions_path = var.datacommons_services_config.instructions_path + artifacts_bucket_name = module.storage.artifacts_bucket_name + vpc_connector_id = var.redis_config.enable ? module.redis[0].connector_id : null + use_spanner = var.spanner_config.enable + env_vars = local.cloud_run_shared_env_variables + secret_env_vars = local.datacommons_services_secrets resolve_with_spanner_embeddings = var.datacommons_services_config.resolve_with_spanner_embeddings depends_on = [module.ingestion_preprocessing_job] @@ -297,6 +297,13 @@ resource "google_storage_bucket_iam_member" "preprocessing_bucket_access" { member = "serviceAccount:${module.ingestion_preprocessing_job[0].service_account_email}" } +resource "google_project_iam_member" "workflow_invoker" { + count = var.ingestion_config.enable_ingestion ? 1 : 0 + project = var.global.project_id + role = "roles/workflows.invoker" + member = "serviceAccount:${module.ingestion_workflow.service_account_email}" +} + resource "google_cloud_run_v2_job_iam_member" "workflow_pre_viewer" { count = var.ingestion_config.enable_ingestion ? 1 : 0 location = var.global.region @@ -313,14 +320,6 @@ resource "google_cloud_run_v2_job_iam_member" "workflow_pre_invoker" { member = "serviceAccount:${module.ingestion_workflow.service_account_email}" } -resource "google_cloud_run_v2_job_iam_member" "workflow_pre_developer" { - count = var.ingestion_config.enable_ingestion ? 1 : 0 - location = var.global.region - name = module.ingestion_preprocessing_job[0].job_name - role = "roles/run.developer" - member = "serviceAccount:${module.ingestion_workflow.service_account_email}" -} - resource "google_project_iam_member" "workflow_dataflow_developer" { count = var.ingestion_config.enable_ingestion ? 1 : 0 project = var.global.project_id From 2b4c38cfd1e28f92fa51efa6e0e7492a0f806519 Mon Sep 17 00:00:00 2001 From: Gabriel Mechali Date: Wed, 8 Jul 2026 17:57:19 -0400 Subject: [PATCH 02/10] Break the dependency between preproc and workflow --- infra/dcp/modules/ingestion/workflow/main.tf | 1 + infra/dcp/modules/ingestion/workflow/variables.tf | 6 ++++++ infra/dcp/modules/ingestion/workflow/workflow.yaml | 2 +- infra/dcp/modules/stack/main.tf | 13 +++++++------ 4 files changed, 15 insertions(+), 7 deletions(-) diff --git a/infra/dcp/modules/ingestion/workflow/main.tf b/infra/dcp/modules/ingestion/workflow/main.tf index 52cf5a4a..14a2d150 100644 --- a/infra/dcp/modules/ingestion/workflow/main.tf +++ b/infra/dcp/modules/ingestion/workflow/main.tf @@ -32,6 +32,7 @@ resource "google_workflows_workflow" "ingestion_orchestrator" { embeddings_timeout = var.embeddings_timeout clean_namespace_prefix = local.clean_namespace_prefix enable_redis_cache_clearing = var.enable_redis_cache_clearing + preprocessing_job_name = var.preprocessing_job_name }) } diff --git a/infra/dcp/modules/ingestion/workflow/variables.tf b/infra/dcp/modules/ingestion/workflow/variables.tf index 12f50638..3a097d67 100644 --- a/infra/dcp/modules/ingestion/workflow/variables.tf +++ b/infra/dcp/modules/ingestion/workflow/variables.tf @@ -59,6 +59,12 @@ variable "enable_redis_cache_clearing" { default = false } +variable "preprocessing_job_name" { + type = string + description = "Name of the ingestion preprocessing Cloud Run job" + default = "" +} + variable "ingestion_artifacts_path" { type = string description = "Path where pre-processed files are placed for the next stage" diff --git a/infra/dcp/modules/ingestion/workflow/workflow.yaml b/infra/dcp/modules/ingestion/workflow/workflow.yaml index 55c9c2ab..f3ed6776 100644 --- a/infra/dcp/modules/ingestion/workflow/workflow.yaml +++ b/infra/dcp/modules/ingestion/workflow/workflow.yaml @@ -20,7 +20,7 @@ main: - decoded_imports: '$${json.decode(input.importList)}' - num_imports: '$${len(decoded_imports)}' - combined_import_name: "" - - prep_job_name: '${clean_namespace_prefix}dc-ingestion-preprocessing-job' + - prep_job_name: '${preprocessing_job_name}' - comma_separated_imports: "" - imports_status_list: [] - imports_history_list: [] diff --git a/infra/dcp/modules/stack/main.tf b/infra/dcp/modules/stack/main.tf index 9dcac086..694c0d34 100644 --- a/infra/dcp/modules/stack/main.tf +++ b/infra/dcp/modules/stack/main.tf @@ -38,11 +38,6 @@ locals { # Use index [0] because module.spanner is now conditional (count). Fallback to empty string if disabled. value = var.spanner_config.enable ? module.spanner[0].spanner_database_id : "" }, - { - name = "INGESTION_WORKFLOW_NAME" - # Fallback to empty string if ingestion is disabled and module output is null - value = module.ingestion_workflow.workflow_name != null ? module.ingestion_workflow.workflow_name : "" - }, { name = "TEMP_LOCATION" value = "gs://${module.storage.artifacts_bucket_name}/${var.ingestion_config.ingestion_artifacts_path}/temp" @@ -198,6 +193,7 @@ module "ingestion_workflow" { dataflow_ip_configuration = var.ingestion_config.dataflow_ip_configuration dataflow_subnetwork = var.ingestion_config.dataflow_subnetwork dataflow_template_gcs_path = var.ingestion_config.dataflow_template_gcs_path + preprocessing_job_name = var.ingestion_config.enable_ingestion ? module.ingestion_preprocessing_job[0].job_name : "" depends_on = [module.ingestion_helper_service] } @@ -250,7 +246,12 @@ module "datacommons_services" { artifacts_bucket_name = module.storage.artifacts_bucket_name vpc_connector_id = var.redis_config.enable ? module.redis[0].connector_id : null use_spanner = var.spanner_config.enable - env_vars = local.cloud_run_shared_env_variables + env_vars = concat(local.cloud_run_shared_env_variables, [ + { + name = "INGESTION_WORKFLOW_NAME" + value = module.ingestion_workflow.workflow_name != null ? module.ingestion_workflow.workflow_name : "" + } + ]) secret_env_vars = local.datacommons_services_secrets resolve_with_spanner_embeddings = var.datacommons_services_config.resolve_with_spanner_embeddings From 7a7f125cc9190231fcd762630265ddf73aee8663 Mon Sep 17 00:00:00 2001 From: Gabriel Mechali Date: Wed, 8 Jul 2026 18:20:54 -0400 Subject: [PATCH 03/10] Move the preproc before lock --- .../modules/ingestion/workflow/workflow.yaml | 50 ++++++++++--------- 1 file changed, 27 insertions(+), 23 deletions(-) diff --git a/infra/dcp/modules/ingestion/workflow/workflow.yaml b/infra/dcp/modules/ingestion/workflow/workflow.yaml index f3ed6776..f6a5878f 100644 --- a/infra/dcp/modules/ingestion/workflow/workflow.yaml +++ b/infra/dcp/modules/ingestion/workflow/workflow.yaml @@ -58,6 +58,33 @@ main: - sanitized_short_import: '$${text.substring(combined_import_name, 0, substring_end)}' - dataflow_job_name: '$${"${clean_namespace_prefix}" + sanitized_short_import + "-" + string(int(sys.now()))}' + - run_preprocessing: + call: launch_and_poll_preprocessing_job + args: + job_name: '$${prep_job_name}' + workflow_id: '$${workflow_id}' + temp_location: '$${input.tempLocation}' + imports_arg: '$${comma_separated_imports}' + region: '$${input.region}' + project_id: '${project_id}' + result: prep_res + + - read_handshake_file: + call: googleapis.storage.v1.objects.get + args: + bucket: '$${bucket_name}' + object: '$${"datacommons/ingestion_records/" + workflow_id + ".json"}' + alt: 'media' + result: handshake_data + + - update_launch_params: + assign: + - launch_params.spannerInstanceId: '$${handshake_data.spannerInstanceId}' + - launch_params.spannerDatabaseId: '$${handshake_data.spannerDatabaseId}' + - launch_params.importList: '$${handshake_data.importList}' + - launch_params.tempLocation: '$${handshake_data.tempLocation}' + - launch_params.stagingLocation: '$${handshake_data.tempLocation}' + - try_acquire_lock: try: call: http.post @@ -110,29 +137,6 @@ main: body: imports: '$${imports_status_list}' status: 'STAGING' - - - run_preprocessing: - call: launch_and_poll_preprocessing_job - args: - job_name: '$${prep_job_name}' - workflow_id: '$${workflow_id}' - temp_location: '$${input.tempLocation}' - imports_arg: '$${comma_separated_imports}' - region: '$${input.region}' - project_id: '${project_id}' - result: prep_res - - - read_handshake_file: - call: googleapis.storage.v1.objects.get - args: - bucket: '$${bucket_name}' - object: '$${"datacommons/ingestion_records/" + workflow_id + ".json"}' - alt: 'media' - result: handshake_data - - - update_launch_params: - assign: - - launch_params.importList: '$${handshake_data.importList}' - launch_dataflow: call: launch_and_poll_dataflow From 71f0649083ae3be1b0715a77f37af910f0514b97 Mon Sep 17 00:00:00 2001 From: Gabriel Mechali Date: Wed, 8 Jul 2026 18:31:10 -0400 Subject: [PATCH 04/10] fix the permissions --- .../modules/ingestion/preprocessing_job/main.tf | 8 -------- .../dcp/modules/ingestion/workflow/workflow.yaml | 10 +++++----- infra/dcp/modules/stack/main.tf | 15 +++++++++++++++ 3 files changed, 20 insertions(+), 13 deletions(-) diff --git a/infra/dcp/modules/ingestion/preprocessing_job/main.tf b/infra/dcp/modules/ingestion/preprocessing_job/main.tf index 243c066c..484bf290 100644 --- a/infra/dcp/modules/ingestion/preprocessing_job/main.tf +++ b/infra/dcp/modules/ingestion/preprocessing_job/main.tf @@ -114,11 +114,3 @@ resource "google_secret_manager_secret_iam_member" "preprocessing_maps_key_acces role = "roles/secretmanager.secretAccessor" member = "serviceAccount:${google_service_account.preprocessing_sa.email}" } - -resource "google_project_iam_member" "preprocessing_workflow_invoker" { - project = var.project_id - role = "roles/workflows.invoker" - member = "serviceAccount:${google_service_account.preprocessing_sa.email}" -} - - diff --git a/infra/dcp/modules/ingestion/workflow/workflow.yaml b/infra/dcp/modules/ingestion/workflow/workflow.yaml index f6a5878f..f04db88b 100644 --- a/infra/dcp/modules/ingestion/workflow/workflow.yaml +++ b/infra/dcp/modules/ingestion/workflow/workflow.yaml @@ -17,19 +17,19 @@ main: - max_lock_retries: $${int(lock_timeout / 120)} - run_embeddings: ${enable_embeddings_generation} - run_postproc: ${enable_bigquery_postprocessing} - - decoded_imports: '$${json.decode(input.importList)}' + - decoded_imports: '$${json.decode(default(map.get(input, "importList"), "[]"))}' - num_imports: '$${len(decoded_imports)}' - combined_import_name: "" - prep_job_name: '${preprocessing_job_name}' - - comma_separated_imports: "" + - comma_separated_imports: '$${default(map.get(input, "imports"), "")}' - imports_status_list: [] - imports_history_list: [] - imports_version_list: [] - launch_params: projectId: '$${project_id}' - spannerInstanceId: '$${input.spannerInstanceId}' - spannerDatabaseId: '$${input.spannerDatabaseId}' - importList: '$${input.importList}' + spannerInstanceId: '$${default(map.get(input, "spannerInstanceId"), "")}' + spannerDatabaseId: '$${default(map.get(input, "spannerDatabaseId"), "")}' + importList: '$${default(map.get(input, "importList"), "")}' tempLocation: '$${input.tempLocation}' stagingLocation: '$${input.tempLocation}' forceCombineNodes: 'true' diff --git a/infra/dcp/modules/stack/main.tf b/infra/dcp/modules/stack/main.tf index 694c0d34..59fb87dd 100644 --- a/infra/dcp/modules/stack/main.tf +++ b/infra/dcp/modules/stack/main.tf @@ -321,6 +321,21 @@ resource "google_cloud_run_v2_job_iam_member" "workflow_pre_invoker" { member = "serviceAccount:${module.ingestion_workflow.service_account_email}" } +resource "google_cloud_run_v2_job_iam_member" "workflow_pre_developer" { + count = var.ingestion_config.enable_ingestion ? 1 : 0 + location = var.global.region + name = module.ingestion_preprocessing_job[0].job_name + role = "roles/run.developer" + member = "serviceAccount:${module.ingestion_workflow.service_account_email}" +} + +resource "google_service_account_iam_member" "workflow_pre_sa_user" { + count = var.ingestion_config.enable_ingestion ? 1 : 0 + service_account_id = "projects/${var.global.project_id}/serviceAccounts/${module.ingestion_preprocessing_job[0].service_account_email}" + role = "roles/iam.serviceAccountUser" + member = "serviceAccount:${module.ingestion_workflow.service_account_email}" +} + resource "google_project_iam_member" "workflow_dataflow_developer" { count = var.ingestion_config.enable_ingestion ? 1 : 0 project = var.global.project_id From 1b58abe54f01f20275eb202822ae03355eb3df28 Mon Sep 17 00:00:00 2001 From: Gabriel Mechali Date: Wed, 8 Jul 2026 19:16:40 -0400 Subject: [PATCH 05/10] Fix the handshake path --- infra/dcp/modules/ingestion/workflow/workflow.yaml | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/infra/dcp/modules/ingestion/workflow/workflow.yaml b/infra/dcp/modules/ingestion/workflow/workflow.yaml index f04db88b..234eb317 100644 --- a/infra/dcp/modules/ingestion/workflow/workflow.yaml +++ b/infra/dcp/modules/ingestion/workflow/workflow.yaml @@ -7,6 +7,7 @@ main: - workflow_id: '$${sys.get_env("GOOGLE_CLOUD_WORKFLOW_EXECUTION_ID")}' - version: '$${"version-" + string(int(sys.now()))}' - bucket_name: '$${text.split(input.tempLocation, "/")[2]}' + - temp_path_prefix: '$${text.replace_all(input.tempLocation, "gs://" + bucket_name + "/", "")}' - execution_error: null - postprocessing_result: null - embedding_result: null @@ -73,7 +74,7 @@ main: call: googleapis.storage.v1.objects.get args: bucket: '$${bucket_name}' - object: '$${"datacommons/ingestion_records/" + workflow_id + ".json"}' + object: '$${temp_path_prefix + "/datacommons/ingestion_records/" + workflow_id + ".json"}' alt: 'media' result: handshake_data From eb29c737a133a08c269c5196c9db82d2ca620a3d Mon Sep 17 00:00:00 2001 From: Gabriel Mechali Date: Wed, 8 Jul 2026 19:24:45 -0400 Subject: [PATCH 06/10] Fix the handshake path --- infra/dcp/modules/ingestion/workflow/workflow.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/infra/dcp/modules/ingestion/workflow/workflow.yaml b/infra/dcp/modules/ingestion/workflow/workflow.yaml index 234eb317..db3d7b72 100644 --- a/infra/dcp/modules/ingestion/workflow/workflow.yaml +++ b/infra/dcp/modules/ingestion/workflow/workflow.yaml @@ -74,7 +74,7 @@ main: call: googleapis.storage.v1.objects.get args: bucket: '$${bucket_name}' - object: '$${temp_path_prefix + "/datacommons/ingestion_records/" + workflow_id + ".json"}' + object: '$${text.url_encode(temp_path_prefix + "/datacommons/ingestion_records/" + workflow_id + ".json")}' alt: 'media' result: handshake_data From 8492601e389c78a8fe9649660d6c9ff1636ef797 Mon Sep 17 00:00:00 2001 From: Gabriel Mechali Date: Wed, 8 Jul 2026 22:31:37 -0400 Subject: [PATCH 07/10] Update CLI --- .../modules/ingestion/workflow/workflow.yaml | 30 ++++++- .../datacommons_admin/ingest_cli.py | 47 ++++------ .../datacommons_admin/ingestion_job_client.py | 82 +++++++++++++---- .../datacommons-admin/tests/test_admin_cli.py | 90 ++++++++++++++----- 4 files changed, 173 insertions(+), 76 deletions(-) diff --git a/infra/dcp/modules/ingestion/workflow/workflow.yaml b/infra/dcp/modules/ingestion/workflow/workflow.yaml index db3d7b72..9dfd732c 100644 --- a/infra/dcp/modules/ingestion/workflow/workflow.yaml +++ b/infra/dcp/modules/ingestion/workflow/workflow.yaml @@ -18,11 +18,11 @@ main: - max_lock_retries: $${int(lock_timeout / 120)} - run_embeddings: ${enable_embeddings_generation} - run_postproc: ${enable_bigquery_postprocessing} - - decoded_imports: '$${json.decode(default(map.get(input, "importList"), "[]"))}' - - num_imports: '$${len(decoded_imports)}' + - decoded_imports: [] + - num_imports: 0 - combined_import_name: "" - prep_job_name: '${preprocessing_job_name}' - - comma_separated_imports: '$${default(map.get(input, "imports"), "")}' + - comma_separated_imports: "" - imports_status_list: [] - imports_history_list: [] - imports_version_list: [] @@ -35,6 +35,30 @@ main: stagingLocation: '$${input.tempLocation}' forceCombineNodes: 'true' isBaseDc: 'false' + - normalize_imports: + switch: + - condition: '$${default(map.get(input, "importList"), "") != ""}' + assign: + - decoded_imports: '$${json.decode(input.importList)}' + - condition: '$${default(map.get(input, "imports"), []) != []}' + steps: + - init_decoded: + assign: + - decoded_imports: [] + - build_decoded_loop: + for: + value: imp_str + in: '$${input.imports}' + steps: + - add_to_decoded: + assign: + - decoded_imports: '$${list.concat(decoded_imports, {"importName": imp_str, "graphPath": ""})}' + - condition: true + assign: + - decoded_imports: [] + - set_num_imports: + assign: + - num_imports: '$${len(decoded_imports)}' - build_lists: for: value: imp diff --git a/packages/datacommons-admin/datacommons_admin/ingest_cli.py b/packages/datacommons-admin/datacommons_admin/ingest_cli.py index 08c04a24..037852f7 100644 --- a/packages/datacommons-admin/datacommons_admin/ingest_cli.py +++ b/packages/datacommons-admin/datacommons_admin/ingest_cli.py @@ -51,54 +51,41 @@ def start(imports: str | None = None) -> None: region = get_region() workflow_name = get_ingestion_workflow_name() - click.secho(f"Found data job: {job_name}", fg="green") + click.secho(f"Found workflow: {workflow_name}", fg="green") click.secho(f"Found workflow service account: {sa_email}", fg="green") click.secho(f"Found GCP project ID: {project_id}", fg="green") click.secho(f"Found GCP region: {region}", fg="green") click.secho( - f"Starting Cloud Run job '{job_name}' via Admin API (this may take a few moments)...", + f"Starting Cloud Workflow '{workflow_name}' via Executions API (this may take a few moments)...", fg="bright_black", ) client = IngestionJobClient( - job_name, + workflow_name=workflow_name, + job_name=job_name, service_account_email=sa_email, project_id=project_id, location=region, ) - result = client.start_job(imports=imports) + result = client.start_workflow(imports=imports) - click.secho("Successfully started ingestion job!", fg="green", bold=True) - res_name = result.get("name") or result.get("metadata", {}).get("name") + click.secho("Successfully started ingestion workflow!", fg="green", bold=True) + res_name = result.get("name") if res_name: - op_pattern = r"projects/([^/]+)/locations/([^/]+)/operations/([^/]+)" - op_match = re.match(op_pattern, res_name) - - if op_match: - click.secho(f"Operation details: {res_name}", fg="bright_black") - resp_project_id, location, operation_id = op_match.groups() + exec_pattern = r"projects/([^/]+)/locations/([^/]+)/workflows/([^/]+)/executions/([^/]+)" + exec_match = re.match(exec_pattern, res_name) - short_job_name = job_name.split("/")[-1] if "/" in job_name else job_name - job_url = f"https://console.cloud.google.com/run/jobs/details/{location}/{short_job_name}/executions?project={resp_project_id}" + if exec_match: + resp_project_id, location, wf_name, exec_id = exec_match.groups() + execution_url = f"https://console.cloud.google.com/workflows/workflow/{location}/{wf_name}/executions/view/{exec_id}?project={resp_project_id}" - click.secho("Operation ID: ", fg="cyan", bold=True, nl=False) - click.secho(operation_id, fg="green") - click.secho("Job console link: ", fg="cyan", bold=True, nl=False) - click.secho(job_url, fg="blue", underline=True) + click.secho("Execution ID: ", fg="cyan", bold=True, nl=False) + click.secho(exec_id, fg="green") + click.secho("Execution console link: ", fg="cyan", bold=True, nl=False) + click.secho(execution_url, fg="blue", underline=True) else: - click.secho(f"Resource details: {res_name}", fg="bright_black") - - click.secho("\n[!] Note on ingestion completion", fg="yellow", bold=True) - click.secho( - "This job triggers a Cloud Workflow that runs in the background.\n" - "Check the Workflows console below to verify full completion.", - fg="yellow", - ) - - workflow_url = f"https://console.cloud.google.com/workflows/workflow/{region}/{workflow_name}/executions?project={project_id}" - click.secho("Workflow Console Link: ", fg="cyan", bold=True, nl=False) - click.secho(workflow_url, fg="blue", underline=True) + click.secho(f"Execution resource path: {res_name}", fg="bright_black") @ingest.command(name="show-config") diff --git a/packages/datacommons-admin/datacommons_admin/ingestion_job_client.py b/packages/datacommons-admin/datacommons_admin/ingestion_job_client.py index 93f9c4d7..bf2649c4 100644 --- a/packages/datacommons-admin/datacommons_admin/ingestion_job_client.py +++ b/packages/datacommons-admin/datacommons_admin/ingestion_job_client.py @@ -12,33 +12,47 @@ # See the License for the specific language governing permissions and # limitations under the License. +import json import click import google.auth from google.auth.transport.requests import AuthorizedSession class IngestionJobClient: - """Client for interacting with Cloud Run Admin API to manage CDC data ingestion jobs.""" + """Client for interacting with Cloud Workflows and Cloud Run Admin APIs to manage CDC data ingestion.""" def __init__( self, job_name: str, + workflow_name: str = None, service_account_email: str = None, project_id: str = None, location: str = None, ) -> None: self.service_account_email = service_account_email + self.project_id = project_id + self.location = location base_credentials, _ = google.auth.default() - if not job_name.startswith("projects/"): - if not project_id: - raise click.ClickException( - "Project ID must be provided via Terraform outputs or as an argument when job name is not a full resource name." - ) - if not location: - raise click.ClickException( - "Location must be provided via Terraform outputs or as an argument when job name is not a full resource name." + if workflow_name: + if not workflow_name.startswith("projects/"): + if not project_id: + raise click.ClickException( + "Project ID must be provided via Terraform outputs or as an argument when workflow name is not a full resource name." + ) + if not location: + raise click.ClickException( + "Location must be provided via Terraform outputs or as an argument when workflow name is not a full resource name." + ) + self.full_workflow_name = ( + f"projects/{project_id}/locations/{location}/workflows/{workflow_name}" ) + else: + self.full_workflow_name = workflow_name + else: + self.full_workflow_name = None + + if not job_name.startswith("projects/"): self.full_job_name = ( f"projects/{project_id}/locations/{location}/jobs/{job_name}" ) @@ -58,27 +72,57 @@ def __init__( self.session = AuthorizedSession(creds) - def start_job(self, imports: str | None = None) -> dict: - """Starts an execution of the Cloud Run job.""" - url = f"https://run.googleapis.com/v2/{self.full_job_name}:run" + def start_workflow(self, imports: str | None = None) -> dict: + """Starts an execution of the Cloud Workflow.""" + if not self.full_workflow_name: + raise click.ClickException( + "Workflow name must be provided to start a workflow execution." + ) - container_override = {"env": [{"name": "DATA_RUN_MODE", "value": "dcpbridge"}]} - if imports: - container_override["args"] = [f"--imports={imports}"] + # 1. Fetch Cloud Run job configuration to get default bucket, region, etc. + env_vars = self.get_config() + env_dict = {env["name"]: env.get("value") for env in env_vars if "name" in env} - json_payload = {"overrides": {"containerOverrides": [container_override]}} + temp_location = env_dict.get("TEMP_LOCATION") + spanner_instance = env_dict.get("GCP_SPANNER_INSTANCE_ID") + spanner_database = env_dict.get("GCP_SPANNER_DATABASE_NAME") + region = env_dict.get("REGION", self.location) + + if not temp_location: + raise click.ClickException( + "TEMP_LOCATION not found in preprocessing job environment configuration." + ) + + # 2. Parse imports argument + imports_list = [] + if imports: + imports_list = [imp.strip() for imp in imports.split(",") if imp.strip()] + + # 3. Construct payload argument (must be a JSON string) + argument_dict = { + "tempLocation": temp_location, + "spannerInstanceId": spanner_instance or "", + "spannerDatabaseId": spanner_database or "", + "region": region, + "imports": imports_list + } + + url = f"https://workflowexecutions.googleapis.com/v1/{self.full_workflow_name}/executions" + json_payload = { + "argument": json.dumps(argument_dict) + } try: response = self.session.post(url, json=json_payload, timeout=300) except Exception as e: - msg = f"Network or authentication error connecting to Cloud Run Admin API at {url}: {e}" + msg = f"Network or authentication error connecting to Workflow Executions API at {url}: {e}" if self.service_account_email: msg += f"\nFailed to impersonate {self.service_account_email}. Please ensure your GCP user account has the 'Service Account Token Creator' (roles/iam.serviceAccountTokenCreator) IAM role." raise click.ClickException(msg) if response.status_code == 401: raise click.ClickException( - f"HTTP 401 Unauthorized when calling Cloud Run Admin API at {url}.\n" + f"HTTP 401 Unauthorized when calling Workflow Executions API at {url}.\n" "Your GCP credentials were rejected. Please verify your authentication.\n" "To re-authenticate, run:\n" " gcloud auth application-default login" @@ -96,7 +140,7 @@ def start_job(self, imports: str | None = None) -> dict: error_msg = response.text raise click.ClickException( - f"Cloud Run Admin API returned HTTP {response.status_code}: {error_msg}" + f"Workflow Executions API returned HTTP {response.status_code}: {error_msg}" ) try: diff --git a/packages/datacommons-admin/tests/test_admin_cli.py b/packages/datacommons-admin/tests/test_admin_cli.py index 946426fc..831fa234 100644 --- a/packages/datacommons-admin/tests/test_admin_cli.py +++ b/packages/datacommons-admin/tests/test_admin_cli.py @@ -345,24 +345,42 @@ def test_ingest_start_success( mock_auth_default.return_value = (mock_creds, "test-project") mock_session_inst = MagicMock() + + # Mock GET for get_config + mock_get_resp = MagicMock() + mock_get_resp.ok = True + mock_get_resp.json.return_value = { + "template": { + "template": { + "containers": [ + { + "env": [ + {"name": "TEMP_LOCATION", "value": "gs://mock-bucket/temp"}, + {"name": "GCP_SPANNER_INSTANCE_ID", "value": "mock-instance"}, + {"name": "GCP_SPANNER_DATABASE_NAME", "value": "mock-db"}, + {"name": "REGION", "value": "us-central1"}, + ] + } + ] + } + } + } + mock_session_inst.get.return_value = mock_get_resp + mock_resp = MagicMock() mock_resp.ok = True mock_resp.json.return_value = { - "name": "projects/mock-proj/locations/us-central1/operations/op-123" + "name": "projects/mock-proj/locations/us-central1/workflows/mock-workflow/executions/exec-123" } mock_session_inst.post.return_value = mock_resp mock_session.return_value = mock_session_inst result = runner.invoke(admin, ["ingest", "start"]) assert result.exit_code == 0 - assert "Successfully started ingestion job!" in result.output - assert ( - "Operation details: projects/mock-proj/locations/us-central1/operations/op-123" - in result.output - ) - assert "Operation ID: op-123" in result.output + assert "Successfully started ingestion workflow!" in result.output + assert "Execution ID: exec-123" in result.output assert ( - "Job console link: https://console.cloud.google.com/run/jobs/details/us-central1/mock-job/executions?project=mock-proj" + "Execution console link: https://console.cloud.google.com/workflows/workflow/us-central1/mock-workflow/executions/view/exec-123?project=mock-proj" in result.output ) @@ -466,28 +484,52 @@ def test_ingest_start_with_imports_success( mock_auth_default.return_value = (mock_creds, "test-project") mock_session_inst = MagicMock() + + # Mock GET for get_config + mock_get_resp = MagicMock() + mock_get_resp.ok = True + mock_get_resp.json.return_value = { + "template": { + "template": { + "containers": [ + { + "env": [ + {"name": "TEMP_LOCATION", "value": "gs://mock-bucket/temp"}, + {"name": "GCP_SPANNER_INSTANCE_ID", "value": "mock-instance"}, + {"name": "GCP_SPANNER_DATABASE_NAME", "value": "mock-db"}, + {"name": "REGION", "value": "us-central1"}, + ] + } + ] + } + } + } + mock_session_inst.get.return_value = mock_get_resp + mock_resp = MagicMock() mock_resp.ok = True mock_resp.json.return_value = { - "name": "projects/mock-proj/locations/us-central1/operations/op-123" + "name": "projects/mock-proj/locations/us-central1/workflows/mock-workflow/executions/exec-123" } mock_session_inst.post.return_value = mock_resp mock_session.return_value = mock_session_inst result = runner.invoke(admin, ["ingest", "start", "--imports", "oecd,doubleup"]) assert result.exit_code == 0 - assert "Successfully started ingestion job!" in result.output - mock_session_inst.post.assert_called_once_with( - "https://run.googleapis.com/v2/projects/mock-proj/locations/us-central1/jobs/mock-job:run", - json={ - "overrides": { - "containerOverrides": [ - { - "env": [{"name": "DATA_RUN_MODE", "value": "dcpbridge"}], - "args": ["--imports=oecd,doubleup"], - } - ] - } - }, - timeout=300, - ) + assert "Successfully started ingestion workflow!" in result.output + + import json + expected_arg = { + "tempLocation": "gs://mock-bucket/temp", + "spannerInstanceId": "mock-instance", + "spannerDatabaseId": "mock-db", + "region": "us-central1", + "imports": ["oecd", "doubleup"] + } + + # Verify the API was called with the correct argument + called_args = mock_session_inst.post.call_args[1] + assert called_args["timeout"] == 300 + called_payload = called_args["json"] + assert "argument" in called_payload + assert json.loads(called_payload["argument"]) == expected_arg From 550e2557ff656eaf0a3def299680db5526b2dc12 Mon Sep 17 00:00:00 2001 From: Gabriel Mechali Date: Wed, 8 Jul 2026 22:36:11 -0400 Subject: [PATCH 08/10] Fix the permission --- infra/dcp/modules/stack/main.tf | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/infra/dcp/modules/stack/main.tf b/infra/dcp/modules/stack/main.tf index 59fb87dd..1d730151 100644 --- a/infra/dcp/modules/stack/main.tf +++ b/infra/dcp/modules/stack/main.tf @@ -342,3 +342,10 @@ resource "google_project_iam_member" "workflow_dataflow_developer" { role = "roles/dataflow.developer" member = "serviceAccount:${module.ingestion_workflow.service_account_email}" } + +resource "google_project_iam_member" "workflow_run_viewer" { + count = var.ingestion_config.enable_ingestion ? 1 : 0 + project = var.global.project_id + role = "roles/run.viewer" + member = "serviceAccount:${module.ingestion_workflow.service_account_email}" +} From 0ec9c94c9b3c8ecae3d88e8875a81306e139f756 Mon Sep 17 00:00:00 2001 From: Gabriel Mechali Date: Wed, 8 Jul 2026 22:40:47 -0400 Subject: [PATCH 09/10] Fix the workflow --- infra/dcp/modules/ingestion/workflow/workflow.yaml | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/infra/dcp/modules/ingestion/workflow/workflow.yaml b/infra/dcp/modules/ingestion/workflow/workflow.yaml index b2a418ee..12d40d65 100644 --- a/infra/dcp/modules/ingestion/workflow/workflow.yaml +++ b/infra/dcp/modules/ingestion/workflow/workflow.yaml @@ -52,7 +52,10 @@ main: steps: - add_to_decoded: assign: - - decoded_imports: '$${list.concat(decoded_imports, {"importName": imp_str, "graphPath": ""})}' + - new_item: + importName: '$${imp_str}' + graphPath: "" + - decoded_imports: '$${list.concat(decoded_imports, new_item)}' - condition: true assign: - decoded_imports: [] From b355d4a75f7c444ed028f2b0632b997d333227f6 Mon Sep 17 00:00:00 2001 From: Gabriel Mechali Date: Wed, 8 Jul 2026 22:45:34 -0400 Subject: [PATCH 10/10] Fix normalize error --- infra/dcp/modules/ingestion/workflow/workflow.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/infra/dcp/modules/ingestion/workflow/workflow.yaml b/infra/dcp/modules/ingestion/workflow/workflow.yaml index 12d40d65..59d769ca 100644 --- a/infra/dcp/modules/ingestion/workflow/workflow.yaml +++ b/infra/dcp/modules/ingestion/workflow/workflow.yaml @@ -40,7 +40,7 @@ main: - condition: '$${default(map.get(input, "importList"), "") != ""}' assign: - decoded_imports: '$${json.decode(input.importList)}' - - condition: '$${default(map.get(input, "imports"), []) != []}' + - condition: '$${len(default(map.get(input, "imports"), [])) > 0}' steps: - init_decoded: assign: