fix: 1308 refresh view cloud task not created when a new dataset is downloaded (#1316)

Author: qcdyx

api/src/shared/common/gcp_utils.py

@@ -34,41 +34,35 @@ def create_refresh_materialized_view_task():
         proto_time.FromDatetime(bucket_time)
 
         # Cloud Tasks setup
-        client = tasks_v2.CloudTasksClient()
         project = os.getenv("PROJECT_ID")
-        location = os.getenv("LOCATION")
         queue = os.getenv("MATERIALIZED_VIEW_QUEUE")
-        url = (
-            f"https://{os.getenv('GCP_REGION')}-"
-            f"{os.getenv('PROJECT_ID')}.cloudfunctions.net/"
-            f"tasks-executor-{os.getenv('ENVIRONMENT_NAME')}"
-        )
-
-        task_name = client.task_path(project, location, queue, task_name)
+        logging.debug("Queue name from env: %s", queue)
+        gcp_region = os.getenv("GCP_REGION")
+        environment_name = os.getenv("ENVIRONMENT")
+        url = f"https://{gcp_region}-" f"{project}.cloudfunctions.net/" f"tasks-executor-{environment_name}"
 
         # Enqueue the task
         try:
             create_http_task_with_name(
-                client=client,
+                client=tasks_v2.CloudTasksClient(),
                 body=b"",
                 url=url,
                 project_id=project,
-                gcp_region=location,
+                gcp_region=gcp_region,
                 queue_name=queue,
                 task_name=task_name,
                 task_time=proto_time,
                 http_method=tasks_v2.HttpMethod.GET,
             )
-            logging.info(f"Scheduled refresh materialized view task for {timestamp_str}")
-            return {"message": f"Refresh task for {timestamp_str} scheduled."}, 200
+            logging.info("Scheduled refresh materialized view task for %s", task_name)
+            return {"message": "Refresh task for %s scheduled." % task_name}, 200
         except Exception as e:
             if "ALREADY_EXISTS" in str(e):
-                logging.info(f"Task already exists for {timestamp_str}, skipping.")
+                logging.info("Task already exists for %s, skipping.", task_name)
 
     except Exception as error:
-        error_msg = f"Error enqueuing task: {error}"
-        logging.error(error_msg)
-        return {"error": error_msg}, 500
+        logging.error("Error enqueuing task: %s", error)
+        return {"error": "Error enqueuing task: %s" % error}, 500
 
 
 def create_http_task_with_name(
@@ -83,11 +77,13 @@ def create_http_task_with_name(
     http_method: "tasks_v2.HttpMethod",
 ):
     """Creates a GCP Cloud Task."""
-
     token = tasks_v2.OidcToken(service_account_email=os.getenv("SERVICE_ACCOUNT_EMAIL"))
 
+    parent = client.queue_path(project_id, gcp_region, queue_name)
+    logging.info("Queue parent path: %s", parent)
+
     task = tasks_v2.Task(
-        name=task_name,
+        name=f"{parent}/tasks/{task_name}",
         schedule_time=task_time,
         http_request=tasks_v2.HttpRequest(
             url=url,
@@ -97,4 +93,10 @@ def create_http_task_with_name(
             headers={"Content-Type": "application/json"},
         ),
     )
-    client.create_task(parent=client.queue_path(project_id, gcp_region, queue_name), task=task)
+    logging.info("Task created with task_name: %s", task_name)
+    try:
+        response = client.create_task(parent=parent, task=task)
+    except Exception as e:
+        logging.error("Error creating task: %s", e)
+        logging.error("response: %s", response)
+    logging.info("Successfully created task in create_http_task_with_name")

functions-python/helpers/utils.py

@@ -182,7 +182,7 @@ def create_http_task(
         project_id=project_id,
         gcp_region=gcp_region,
         queue_name=queue_name,
-        task_name="task_name",
+        task_name=None,  # No specific task name provided
         task_time=proto_time,
         http_method=tasks_v2.HttpMethod.POST,
     )

infra/batch/main.tf

@@ -41,6 +41,8 @@ locals {
   public_hosted_datasets_url = lower(var.environment) == "prod" ? "https://${var.public_hosted_datasets_dns}" : "https://${var.environment}-${var.public_hosted_datasets_dns}"
   # 1day=86400, 7days=604800, 31days=2678400
   retention_duration_seconds = lower(var.environment) == "prod" ? 2678400 : 604800
+
+  deployment_timestamp = formatdate("YYYYMMDDhhmmss", timestamp())
 }
 
 data "google_vpc_access_connector" "vpc_connector" {
@@ -229,6 +231,20 @@ resource "google_project_iam_member" "datastore_owner" {
   member  = "serviceAccount:${google_service_account.functions_service_account.email}"
 }
 
+# Grant the batch functions service account permission to enqueue Cloud Tasks
+resource "google_project_iam_member" "queue_enqueuer" {
+  project = var.project_id
+  role    = "roles/cloudtasks.enqueuer"
+  member  = "serviceAccount:${google_service_account.functions_service_account.email}"
+}
+
+# This permission is added to allow the function to act as the service account and generate tokens.
+resource "google_project_iam_member" "service_account_workflow_act_as_binding" {
+  project = var.project_id
+  role    = "roles/iam.serviceAccountUser" #iam.serviceAccounts.actAs
+  member  = "serviceAccount:${google_service_account.functions_service_account.email}"
+}
+
 resource "google_pubsub_topic" "pubsub_topic" {
   name = "datasets-batch-topic-${var.environment}"
 }
@@ -264,6 +280,11 @@ resource "google_cloudfunctions2_function" "pubsub_function" {
       DB_REUSE_SESSION    = "True"
       ENVIRONMENT         = var.environment
       PUBLIC_HOSTED_DATASETS_URL = local.public_hosted_datasets_url
+      PROJECT_ID = var.project_id
+      GCP_REGION = var.gcp_region
+      SERVICE_ACCOUNT_EMAIL = google_service_account.functions_service_account.email
+      MATERIALIZED_VIEW_QUEUE = google_cloud_tasks_queue.refresh_materialized_view_task_queue.name
+
     }
     dynamic "secret_environment_variables" {
       for_each = local.function_batch_process_dataset_config.secret_environment_variables
@@ -288,6 +309,26 @@ resource "google_cloudfunctions2_function" "pubsub_function" {
   }
 }
 
+# Task queue to invoke refresh_materialized_view 
+resource "google_cloud_tasks_queue" "refresh_materialized_view_task_queue" {
+  project  = var.project_id
+  location = var.gcp_region
+  name     = "refresh-materialized-view-task-queue-${var.environment}-${local.deployment_timestamp}"
+
+  rate_limits {
+    max_concurrent_dispatches = 1
+    max_dispatches_per_second = 0.5
+  }
+
+  retry_config {
+    # ~22 minutes total: 120 + 240 + 480 + 480 = 1320s (initial attempt + 4 retries)
+    max_attempts  = 5
+    min_backoff   = "120s"
+    max_backoff   = "480s"
+    max_doublings = 2
+  }
+}
+
 resource "google_cloudfunctions2_function_iam_member" "pubsub_function_invoker" {
   cloud_function = google_cloudfunctions2_function.pubsub_function.name
   project        = var.project_id
@@ -374,4 +415,4 @@ resource "google_compute_global_forwarding_rule" "files_http_lb_rule_ipv4" {
   port_range            = "443"
   ip_address            = data.google_compute_global_address.files_http_lb_ipv4.address
   load_balancing_scheme = "EXTERNAL_MANAGED"
-}
+}

infra/functions-python/main.tf

@@ -1422,10 +1422,10 @@ resource "google_cloud_tasks_queue" "refresh_materialized_view_task_queue" {
   }
 
   retry_config {
-    # This will make the cloud task retry for ~30 minutes
+    # ~22 minutes total: 120 + 240 + 480 + 480 = 1320s (initial attempt + 4 retries)
     max_attempts  = 5
     min_backoff   = "120s"
-    max_backoff   = "120s"
+    max_backoff   = "480s"
     max_doublings = 2
   }
 }