WIP - Analyst sees that GTFS-RT feeds are updated via workflows

Author: ohrite

iac/cal-itp-data-infra-staging/gtfs-rt-archiver/us/.terraform.lock.hcl

@@ -0,0 +1,16 @@
+provider "google" {
+  project = "cal-itp-data-infra-staging"
+}
+
+terraform {
+  required_providers {
+    google = {
+      version = "~> 6.41.0"
+    }
+  }
+
+  backend "gcs" {
+    bucket = "calitp-staging-gcp-components-tfstate"
+    prefix = "cal-itp-data-infra-staging/gtfs-rt-archiver"
+  }
+}

iac/cal-itp-data-infra-staging/gtfs-rt-archiver/us/provider.tf

@@ -0,0 +1,21 @@
+locals {
+  workflow_path = "${path.module}/../../../../workflows"
+}
+
+data "terraform_remote_state" "iam" {
+  backend = "gcs"
+
+  config = {
+    bucket = "calitp-staging-gcp-components-tfstate"
+    prefix = "cal-itp-data-infra-staging/iam"
+  }
+}
+
+data "terraform_remote_state" "gcs" {
+  backend = "gcs"
+
+  config = {
+    bucket = "calitp-staging-gcp-components-tfstate"
+    prefix = "cal-itp-data-infra-staging/gcs"
+  }
+}

iac/cal-itp-data-infra-staging/gtfs-rt-archiver/us/variables.tf

@@ -0,0 +1,33 @@
+resource "google_workflows_workflow" "gtfs-rt-archiver" {
+  name            = "gtfs-rt-archiver"
+  region          = "us-west2"
+  project         = "cal-itp-data-infra-staging"
+  description     = "GTFS-RT Archiver"
+  service_account = data.terraform_remote_state.iam.outputs.google_service_account_workflow-service-account_email
+  source_contents = templatefile("${local.workflow_path}/gtfs-rt.yaml", { bucket = data.terraform_remote_state.gcs.outputs.google_storage_bucket_calitp-staging-gtfs-rt-raw-v2_name })
+}
+
+resource "google_cloud_scheduler_job" "gtfs-rt-archiver" {
+  project          = "cal-itp-data-infra-staging"
+  name             = "gtfs-rt-archiver"
+  description      = "Cloud Scheduler GTFS-RT Archiver"
+  schedule         = "*/15 * * * *"
+  time_zone        = "America/Los_Angeles"
+  attempt_deadline = "320s"
+  region           = "us-west2"
+
+  http_target {
+    http_method = "POST"
+    uri         = "https://workflowexecutions.googleapis.com/v1/${google_workflows_workflow.gtfs-rt-archiver.id}/executions"
+    body = base64encode(
+      jsonencode({
+        "argument" : "{}",
+        "callLogLevel" : "CALL_LOG_LEVEL_UNSPECIFIED"
+      })
+    )
+
+    oauth_token {
+      service_account_email = data.terraform_remote_state.iam.outputs.google_service_account_workflow-service-account_email
+    }
+  }
+}

iac/cal-itp-data-infra-staging/gtfs-rt-archiver/us/workflow.tf

@@ -190,6 +190,10 @@ output "google_service_account_composer-service-account_name" {
   value = google_service_account.composer-service-account.name
 }
 
+output "google_service_account_workflow-service-account_email" {
+  value = google_service_account.workflow-service-account.email
+}
+
 output "google_service_account_sftp-pod-service-account_id" {
   value = google_service_account.sftp-pod-service-account.id
 }

iac/cal-itp-data-infra-staging/iam/us/outputs.tf

@@ -294,6 +294,19 @@ resource "google_project_iam_member" "composer-service-account" {
   project = "cal-itp-data-infra-staging"
 }
 
+resource "google_project_iam_member" "workflow-service-account" {
+  for_each = toset([
+    "roles/storage.objectUser",
+    "roles/bigquery.dataViewer",
+    "roles/bigquery.jobUser",
+    "roles/secretmanager.secretAccessor",
+    "roles/workflows.invoker"
+  ])
+  role    = each.key
+  member  = "serviceAccount:${google_service_account.workflow-service-account.email}"
+  project = "cal-itp-data-infra-staging"
+}
+
 resource "google_project_iam_member" "ms-entra-id-DOT_DDS_Data_Pipeline_and_Warehouse_Users" {
   for_each = toset([
     "roles/viewer",

iac/cal-itp-data-infra-staging/iam/us/project_iam_member.tf

@@ -58,6 +58,14 @@ resource "google_service_account" "composer-service-account" {
   project      = "cal-itp-data-infra-staging"
 }
 
+resource "google_service_account" "workflow-service-account" {
+  account_id   = "workflow-service-account"
+  description  = "Service account for Workflow"
+  disabled     = "false"
+  display_name = "workflow"
+  project      = "cal-itp-data-infra-staging"
+}
+
 resource "google_service_account" "sftp-pod-service-account" {
   account_id   = "sftp-pod-service-account"
   description  = "Service account for sftp server"

iac/cal-itp-data-infra-staging/iam/us/service_account.tf

@@ -0,0 +1,75 @@
+main:
+  params: [args]
+  steps:
+    - init:
+        assign:
+          - projectId: $${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
+          - limit: $${default(map.get(args, "limit"), "1")}
+          - dataset: staging
+          - table: int_transit_database__gtfs_datasets_dim
+          - bucket: calitp-staging-gtfs-rt-raw-v2
+    - runQuery:
+        call: googleapis.bigquery.v2.jobs.query
+        args:
+          projectId: $${projectId}
+          body:
+            useLegacySql: false
+            useQueryCache: true
+            timeoutMs: 30000
+            query: $${
+              "SELECT pipeline_url, type, base64_url " +
+              "FROM `" + projectId + "." + dataset + "." + table + "` " +
+              "WHERE _is_current = true " +
+              "AND type IN ('service_alerts', 'trip_updates', 'vehicle_positions') " +
+              "LIMIT " + limit
+              }
+        result: queryResult
+    - fetchInterval:
+        parallel:
+          for:
+            value: fetchSecond
+            in: [0, 20, 40]
+            steps:
+              - wait:
+                  call: sys.sleep
+                  args:
+                    seconds: $${fetchSecond}
+              - fetchUrls:
+                  parallel:
+                    for:
+                      value: row
+                      in: $${queryResult.rows}
+                      steps:
+                        - fetchInit:
+                            assign:
+                              - fetchPipelineUrl: $${row.f[0].v}
+                              - fetchType: $${row.f[1].v}
+                              - fetchBase64Url: $${row.f[2].v}
+                              - fetchDate: $${text.split(time.format(sys.now()), "T")[0]}
+                              - fetchTime: $${text.split(time.format(sys.now()), "T")[1]}
+                              - fetchHour: $${text.split(fetchTime, ":")[0]}
+                              - fetchMinute: $${text.split(fetchTime, ":")[1]}
+                              - fetchHourTimestamp: $${fetchDate + "T" + fetchHour + ":00:00+00:00"}
+                              - fetchTimestamp: $${fetchDate + "T" + fetchHour + ":" + fetchMinute + ":" + if(fetchSecond == 0, "00", fetchSecond) + "+00:00"}
+                        - fetchUrl:
+                            call: http.get
+                            args:
+                              url: $${fetchPipelineUrl}
+                              timeout: 20
+                            result: fetchResult
+                        - writeResult:
+                            call: http.post
+                            args:
+                              url: $${"https://storage.googleapis.com/upload/storage/v1/b/" + bucket + "/o"}
+                              auth:
+                                type: OAuth2
+                              query:
+                                name: $${
+                                  fetchType +
+                                  "/dt=" + fetchDate +
+                                  "/hour=" + fetchHourTimestamp +
+                                  "/ts=" + fetchTimestamp +
+                                  "/base64_url=" + fetchBase64Url +
+                                  "/feed"
+                                  }
+                              body: $${fetchResult.body}