workflow.yaml

main:
  params: [args]
  steps:
    - init:
        assign:
          - project: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
          - cloud_function: ${default(map.get(args, "cloud_function"), "gaarf")}
          - function_location: ${default(map.get(args, "location"), "us-central1")}
          - bq_dataset_location: ${if(default(map.get(args, "bq_dataset_location"), "us") == "", "us", default(map.get(args, "bq_dataset_location"), "us"))}
          - gcs_bucket: ${default(map.get(args,"gcs_bucket"), project)}
          - only_run_bq: ${default(map.get(args,"only_run_bq"), false)}
          - completion_topic: ${default(map.get(args, "completion_topic"), "gaarf_wf_completed")}
          # 20 is the default for "Concurrent branches and iterations" - https://cloud.google.com/workflows/quotas#parallel_steps
          - concurrency_limit: ${default(map.get(args, "concurrency_limit"), 20)}
          - workflow_ads_id: ${default(map.get(args, "workflow_ads_id"), sys.get_env("GOOGLE_CLOUD_WORKFLOW_ID") + "-ads")}
          #- recreate_dataset: ${default(map.get(args,"recreate_dataset"), false)}
          - disable_strict_views: ${default(map.get(args,"disable_strict_views"), false)}
    - check_shortcut_arg:
        switch:
          - condition: ${only_run_bq}
            next: run_bq_workflow
        next: run_ads_workflow
    - run_ads_workflow:
        call: runAdsQueries
        args:
          project: ${project}
          function_location: ${function_location}
          function_name: ${cloud_function}
          gcs_bucket: ${gcs_bucket}
          queries_path: ${args.ads_queries_path}
          ads_config_path: ${args.ads_config_path}
          cid: ${args.cid}
          cid_ignore: ${map.get(args, "cid_ignore")}
          customer_ids_query: ${map.get(args, "customer_ids_query")}
          customer_ids_offset: ${map.get(args, "customer_ids_offset")}
          customer_ids_batchsize: ${map.get(args, "customer_ids_batchsize")}
          writer: ${default(map.get(args, "writer"), "bq")}
          bq_dataset: ${default(map.get(args,"dataset"), "")}
          bq_dataset_location: ${bq_dataset_location}
          macros: ${map.get(args, "ads_macro")}
          writer_options: ${default(map.get(args, "writer_options"), map.get(args, "bq_writer_options"))}
          output_path: ${map.get(args, "output_path")}
          concurrency_limit: ${concurrency_limit}
          workflow_ads_id: ${workflow_ads_id}
          disable_strict_views: ${disable_strict_views}
        result: accounts
    - run_bq_workflow:
        call: runBigQueryQueries
        args:
          project: ${project}
          function_name: ${cloud_function + "-bq"}
          function_location: ${function_location}
          gcs_bucket: ${gcs_bucket}
          queries_path: ${args.bq_queries_path}
          dataset_location: ${bq_dataset_location}
          macros: ${map.get(args, "bq_macro")}
          sqlParams: ${map.get(args, "bq_sql")}
    - create_completion_message:
        assign:
          - message:
              accounts: ${len(accounts)}
          - base64Msg: ${base64.encode(json.encode(message))}
    - publish_completion_message:
        call: googleapis.pubsub.v1.projects.topics.publish
        args:
          topic: ${"projects/" + project + "/topics/" + completion_topic}
          body:
            messages:
              - data: ${base64Msg}
        result: publishResult
    - return_result:
        return: ${accounts}

runAdsQueries:
  params:
    [
      project,
      function_name,
      function_location,
      gcs_bucket,
      queries_path,
      ads_config_path,
      cid,
      cid_ignore,
      customer_ids_query,
      customer_ids_offset,
      customer_ids_batchsize,
      writer,
      bq_dataset,
      bq_dataset_location,
      macros,
      writer_options,
      output_path,
      concurrency_limit,
      workflow_ads_id,
      disable_strict_views,
    ]
  # NOTE: currently it's assumed that CF's project is the same as project for BQ datasets
  steps:
    # get CF 'gaarf-getcids' function's URL
    - get_function_cids:
        call: http.get
        args:
          url: ${"https://cloudfunctions.googleapis.com/v2/projects/" + project + "/locations/" + function_location + "/functions/" + function_name + "-getcids"}
          auth:
            type: OAuth2
        result: function_cids
    # get CF 'gaarf-bq-view' function's URL
    - get_function_view:
        call: http.get
        args:
          url: ${"https://cloudfunctions.googleapis.com/v2/projects/" + project + "/locations/" + function_location + "/functions/" + function_name + "-bq-view"}
          auth:
            type: OAuth2
        result: function_view

    #call 'gaarf-getcids' CF to get a list of customer ids for further processing
    - call_gaarf_cids_cf:
        call: http.post
        args:
          url: ${function_cids.body.serviceConfig.uri}
          timeout: 1800 # maximum allowed timeout in Workflows is 1800 (even though CF gen2 support 3600)
          query:
            ads_config_path: ${ads_config_path}
            customer_id: ${cid}
            customer_ids_ignore: ${default(cid_ignore, "")}
            customer_ids_query: ${default(customer_ids_query, "")}
            customer_ids_offset: ${default(customer_ids_offset, "")}
            customer_ids_batchsize: ${default(customer_ids_batchsize, "")}
          auth:
            type: OIDC
        result: accounts_response
    - set_accounts_from_cf_response:
        assign:
          - accounts: ${accounts_response.body.accounts}
    - log_cids:
        call: sys.log
        args:
          json:
            text: "accounts to process"
            batches: ${len(accounts)}
            accounts: ${accounts}
          severity: "INFO"
    # fetch script from GCS
    - get_ads_scripts_from_gcs:
        call: googleapis.storage.v1.objects.list
        args:
          bucket: ${gcs_bucket}
          prefix: ${queries_path}
        result: scripts_raw

    - initialize_filtered_list:
        assign:
          - scripts: []

    - filter_sql_files:
        for:
          value: script_item
          in: ${scripts_raw.items}
          steps:
            - check_sql_extension:
                switch:
                  - condition: ${text.match_regex(script_item.name, "[.]sql$")}
                    assign:
                      - scripts: ${list.concat(scripts, "gs://" + script_item.bucket + "/" + script_item.name)}

    # now `scripts` is a list of GCS uris of SQL scripts
    - log_ads_scripts:
        call: sys.log
        args:
          json:
            count: ${len(scripts)}
            scripts: ${scripts}
          severity: "INFO"

    # loop over accounts grouped by batches (to overcome the limit of max steps in workflow)
    - iteration_over_batches:
        for:
          value: accounts_batch
          in: ${accounts}
          index: index
          steps:
            - log_cids_batch:
                call: sys.log
                args:
                  json:
                    text: "Executing Ads subworkflow"
                    index: ${index}
                    count: ${len(accounts_batch)}
                    accounts: ${accounts_batch}
                  severity: "INFO"
            - run_ads_workflow:
                call: googleapis.workflowexecutions.v1.projects.locations.workflows.executions.run
                args:
                  workflow_id: ${workflow_ads_id}
                  connector_params:
                    timeout: 86400 # 24h=60*60*24 seconds
                  argument:
                    scripts: ${scripts}
                    accounts: ${accounts_batch}
                    function_name: ${function_name}
                    function_location: ${function_location}
                    ads_config_path: ${ads_config_path}
                    writer: ${writer}
                    bq_dataset: ${bq_dataset}
                    bq_dataset_location: ${bq_dataset_location}
                    macros: ${macros}
                    writer_options: ${writer_options}
                    output_path: ${output_path}
                    concurrency_limit: ${concurrency_limit}
                result: execution_result

    # collecting account ids from `accounts` where they are slit into batched into a flatten list
    - initialize_accounts_flatten:
        assign:
          - accounts_flatten: []
    - flatten_accounts:
        for:
          value: accounts_batch
          in: ${accounts}
          steps:
            - loop_over_batch:
                for:
                  value: account
                  in: ${accounts_batch}
                  steps:
                    - append:
                        assign:
                          - accounts_flatten: ${list.concat(accounts_flatten, account)}

    - check_for_bq_writer:
        switch:
          - condition: ${writer == "bq" or writer == "bigquery"}
            next: create_views
        next: return_result

    # for each script (excluding constants) create a unified view
    - create_views:
        for:
          value: script_item
          in: ${scripts}
          steps:
            - check_for_constant:
                switch:
                  - condition: ${text.match_regex(script_item, "_constant")}
                    next: continue
                next: call_create_view_cf
            - call_create_view_cf:
                call: http.post
                args:
                  url: ${function_view.body.serviceConfig.uri}
                  timeout: 1800
                  query:
                    project_id: ${project}
                    dataset: ${bq_dataset}
                    dataset_location: ${bq_dataset_location}
                    script_path: ${script_item}
                  body:
                    accounts: ${if(disable_strict_views, [], accounts_flatten)}
                  auth:
                    type: OIDC
                result: create_view_response

    - return_result:
        return: ${accounts}

runBigQueryQueries:
  params:
    [
      project,
      function_location,
      function_name,
      gcs_bucket,
      queries_path,
      macros,
      sqlParams,
      dataset_location,
    ]
  steps:
    - get_bq_scripts_from_gcs:
        call: googleapis.storage.v1.objects.list
        args:
          bucket: ${gcs_bucket}
          prefix: ${queries_path}
        result: bq_scripts_raw

    # check if there are any bq scripts on GCS
    - check_scripts:
        switch:
          - condition: ${map.get(bq_scripts_raw, "items") != null and len(map.get(bq_scripts_raw, "items")) > 0}
            next: initialize_filtered_list
        next: end

    - initialize_filtered_list:
        assign:
          - bq_scripts: []

    - filter_sql_files:
        for:
          value: bq_script_item
          in: ${bq_scripts_raw.items}
          steps:
            - check_sql_extension:
                switch:
                  - condition: ${text.match_regex(bq_script_item.name, "[.]sql$")}
                    assign:
                      - bq_scripts: ${list.concat(bq_scripts, bq_script_item.name)}

    - log_bq_scripts:
        call: sys.log
        args:
          data: ${bq_scripts}
          severity: "INFO"

    # check if there are any bq scripts on GCS
    - check_scripts2:
        switch:
          - condition: ${len(bq_scripts) > 0}
            next: get_function_bq
        next: end

    # get clound function's uri
    - get_function_bq:
        call: http.get
        args:
          url: ${"https://cloudfunctions.googleapis.com/v2/projects/" + project + "/locations/" + function_location + "/functions/" + function_name}
          auth:
            type: OAuth2
        result: function_bq
        # TODO: move to using CF adapter when it's ready (currently only v1 supported):
        #call: googleapis.cloudfunctions.v2.projects.locations.functions.get
        #args:
        #  name: ${"projects/" + project + "/locations/" + function_location + "/functions/" + function_name}
        #result: function_bq
    - runBqQueries:
        for:
          value: bq_script_item
          in: ${bq_scripts}
          steps:
            - call_gaarf_bq_cf:
                try:
                  call: http.post
                  args:
                    url: ${function_bq.body.serviceConfig.uri}
                    timeout: 1800
                    query:
                      script_path: ${"gs://" + gcs_bucket + "/" + bq_script_item}
                      project_id: ${project}
                      dataset_location: ${dataset_location}
                    body:
                      macro: ${macros}
                      sql: ${sqlParams}
                    auth:
                      type: OIDC
                  result: script_results
                retry:
                  predicate: ${custom_retry_predicate}
                  max_retries: 3
                  backoff:
                    initial_delay: 2
                    max_delay: 60
                    multiplier: 2
            - log_script_bq_result:
                call: sys.log
                args:
                  data:
                    script: ${bq_script_item}
                    result: ${script_results.body}
                  severity: "INFO"

custom_retry_predicate:
  params: [e]
  steps:
    - log_call_gaarf_cf_failure:
        call: sys.log
        args:
          data: ${e}
          severity: "WARNING"
    - normalize_fields:
        assign:
          - tags: ${default(map.get(e, "tags"), [])}
          - code: ${default(map.get(e, "code"),0)}
    - what_to_repeat:
        switch:
          # We'll repeat if it's a ConnectionError, TimeoutError or http statuses:
          #   429 - Too Many Requests
          #   502 - Bad Gateway
          #   503 - Service Unavailable
          #   504 - Gateway Timeout
          # NOTE: sometime errors happen inside Workflow and there's no any code
          # (i.e. "code" can be null, so DO NOT use operand ==,<,>,>=,<= without wrapping with `default`
          - condition: ${"ConnectionFailedError" in tags or "ConnectionError" in tags or "TimeoutError" in tags or code == 429 or code == 502 or code == 503 or code == 504}
            return: true
    - otherwise:
        return: false