<commit_before># Copyright 2021 The Kubeflow Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Module for scheduling pipeline runs."""

import base64
import hashlib
import json
import logging
import pathlib
import re
import tempfile
from typing import Any, Mapping, Optional
import zipfile

import googleapiclient
from googleapiclient import discovery
import requests

from kfp.v2.google.client import client_utils
from kfp.v2.google.client import runtime_config_builder

_PROXY_FUNCTION_NAME = 'templated_http_request-v1'
_PROXY_FUNCTION_FILENAME = '_cloud_function_templated_http_request.py'

_CAIPP_ENDPOINT_WITHOUT_REGION = 'aiplatform.googleapis.com'
_CAIPP_API_VERSION = 'v1beta1'


def create_from_pipeline_file(
    pipeline_path: str,
    schedule: str,
    project_id: str,
    region: str = 'us-central1',
    time_zone: str = 'US/Pacific',
    parameter_values: Optional[Mapping[str, Any]] = None,
    pipeline_root: Optional[str] = None,
    service_account: Optional[str] = None,
    app_engine_region: Optional[str] = None,
    cloud_scheduler_service_account: Optional[str] = None,
) -> dict:
    """Creates schedule for compiled pipeline file.

    This function creates scheduled job which will run the provided pipeline on
    schedule. This is implemented by creating a Google Cloud Scheduler Job.
    The job will be visible in https://console.google.com/cloudscheduler and can
    be paused/resumed and deleted.

    To make the system work, this function also creates a Google Cloud Function
    which acts as an intermediary between the Scheduler and Pipelines. A single
    function is shared between all scheduled jobs.
    The following APIs will be activated automatically:
    * cloudfunctions.googleapis.com
    * cloudscheduler.googleapis.com
    * appengine.googleapis.com

    Args:
      pipeline_path: Path of the compiled pipeline file.
      schedule: Schedule in cron format. Example: "45 * * * *"
      project_id: Google Cloud project ID
      region: Google Cloud compute region. Default is 'us-central1'
      time_zone: Schedule time zone. Default is 'US/Pacific'
      parameter_values: Arguments for the pipeline parameters
      pipeline_root: Optionally the user can override the pipeline root
        specified during the compile time.
      service_account: The service account that the pipeline workload runs as.
      app_engine_region: The region that cloud scheduler job is created in.
      cloud_scheduler_service_account: The service account that Cloud Scheduler job and the proxy cloud function use.
        this should have permission to call AI Platform API and the proxy function.
        If not specified, the functions uses the App Engine default service account.

    Returns:
      Created Google Cloud Scheduler Job object dictionary.
    """
    pipeline_dict = client_utils.load_json(pipeline_path)

    return _create_from_pipeline_dict(
        pipeline_dict=pipeline_dict,
        schedule=schedule,
        project_id=project_id,
        region=region,
        time_zone=time_zone,
        parameter_values=parameter_values,
        pipeline_root=pipeline_root,
        service_account=service_account,
        app_engine_region=app_engine_region,
        cloud_scheduler_service_account=cloud_scheduler_service_account,
    )


def _create_from_pipeline_dict(
    pipeline_dict: dict,
    schedule: str,
    project_id: str,
    region: str = 'us-central1',
    time_zone: str = 'US/Pacific',
    parameter_values: Optional[Mapping[str, Any]] = None,
    pipeline_root: Optional[str] = None,
    service_account: Optional[str] = None,
    app_engine_region: Optional[str] = None,
    cloud_scheduler_service_account: Optional[str] = None,
) -> dict:
    """Creates schedule for compiled pipeline dictionary."""

    _enable_required_apis(project_id=project_id)

    # If appengine region is not provided, use the pipeline region.
    app_engine_region = app_engine_region or region

    proxy_function_url = _get_proxy_cloud_function_endpoint(
        project_id=project_id,
        region=region,
        cloud_scheduler_service_account=cloud_scheduler_service_account,
    )

    if parameter_values or pipeline_root:
        config_builder = runtime_config_builder.RuntimeConfigBuilder.from_job_spec_json(
            pipeline_dict)
        config_builder.update_runtime_parameters(
            parameter_values=parameter_values)
        config_builder.update_pipeline_root(pipeline_root=pipeline_root)
        updated_runtime_config = config_builder.build()
        pipeline_dict['runtimeConfig'] = updated_runtime_config

    # Creating job creation request to get the final request URL
    pipeline_jobs_api_url = f'https://{region}-{_CAIPP_ENDPOINT_WITHOUT_REGION}/{_CAIPP_API_VERSION}/projects/{project_id}/locations/{region}/pipelineJobs'

    # Preparing the request body for the Cloud Function processing
    full_pipeline_name = pipeline_dict.get('name')
    pipeline_display_name = pipeline_dict.get('displayName')
    time_format_suffix = "-{{$.scheduledTime.strftime('%Y-%m-%d-%H-%M-%S')}}"
    if 'name' in pipeline_dict:
        pipeline_dict['name'] += time_format_suffix
    if 'displayName' in pipeline_dict:
        pipeline_dict['displayName'] += time_format_suffix<commit_msg>fix(sdk): fix cloud scheduler's job name (#6844)<commit_after># Copyright 2021 The Kubeflow Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Module for scheduling pipeline runs."""

import base64
import hashlib
import json
import logging
import pathlib
import re
import tempfile
from typing import Any, Mapping, Optional
import zipfile

import googleapiclient
from googleapiclient import discovery
import requests

from kfp.v2.google.client import client_utils
from kfp.v2.google.client import runtime_config_builder

_PROXY_FUNCTION_NAME = 'templated_http_request-v1'
_PROXY_FUNCTION_FILENAME = '_cloud_function_templated_http_request.py'

_CAIPP_ENDPOINT_WITHOUT_REGION = 'aiplatform.googleapis.com'
_CAIPP_API_VERSION = 'v1beta1'


def create_from_pipeline_file(
    pipeline_path: str,
    schedule: str,
    project_id: str,
    region: str = 'us-central1',
    time_zone: str = 'US/Pacific',
    parameter_values: Optional[Mapping[str, Any]] = None,
    pipeline_root: Optional[str] = None,
    service_account: Optional[str] = None,
    app_engine_region: Optional[str] = None,
    cloud_scheduler_service_account: Optional[str] = None,
) -> dict:
    """Creates schedule for compiled pipeline file.

    This function creates scheduled job which will run the provided pipeline on
    schedule. This is implemented by creating a Google Cloud Scheduler Job.
    The job will be visible in https://console.google.com/cloudscheduler and can
    be paused/resumed and deleted.

    To make the system work, this function also creates a Google Cloud Function
    which acts as an intermediary between the Scheduler and Pipelines. A single
    function is shared between all scheduled jobs.
    The following APIs will be activated automatically:
    * cloudfunctions.googleapis.com
    * cloudscheduler.googleapis.com
    * appengine.googleapis.com

    Args:
      pipeline_path: Path of the compiled pipeline file.
      schedule: Schedule in cron format. Example: "45 * * * *"
      project_id: Google Cloud project ID
      region: Google Cloud compute region. Default is 'us-central1'
      time_zone: Schedule time zone. Default is 'US/Pacific'
      parameter_values: Arguments for the pipeline parameters
      pipeline_root: Optionally the user can override the pipeline root
        specified during the compile time.
      service_account: The service account that the pipeline workload runs as.
      app_engine_region: The region that cloud scheduler job is created in.
      cloud_scheduler_service_account: The service account that Cloud Scheduler job and the proxy cloud function use.
        this should have permission to call AI Platform API and the proxy function.
        If not specified, the functions uses the App Engine default service account.

    Returns:
      Created Google Cloud Scheduler Job object dictionary.
    """
    pipeline_dict = client_utils.load_json(pipeline_path)

    return _create_from_pipeline_dict(
        pipeline_dict=pipeline_dict,
        schedule=schedule,
        project_id=project_id,
        region=region,
        time_zone=time_zone,
        parameter_values=parameter_values,
        pipeline_root=pipeline_root,
        service_account=service_account,
        app_engine_region=app_engine_region,
        cloud_scheduler_service_account=cloud_scheduler_service_account,
    )


def _create_from_pipeline_dict(
    pipeline_dict: dict,
    schedule: str,
    project_id: str,
    region: str = 'us-central1',
    time_zone: str = 'US/Pacific',
    parameter_values: Optional[Mapping[str, Any]] = None,
    pipeline_root: Optional[str] = None,
    service_account: Optional[str] = None,
    app_engine_region: Optional[str] = None,
    cloud_scheduler_service_account: Optional[str] = None,
) -> dict:
    """Creates schedule for compiled pipeline dictionary."""

    _enable_required_apis(project_id=project_id)

    # If appengine region is not provided, use the pipeline region.
    app_engine_region = app_engine_region or region

    proxy_function_url = _get_proxy_cloud_function_endpoint(
        project_id=project_id,
        region=region,
        cloud_scheduler_service_account=cloud_scheduler_service_account,
    )

    if parameter_values or pipeline_root:
        config_builder = runtime_config_builder.RuntimeConfigBuilder.from_job_spec_json(
            pipeline_dict)
        config_builder.update_runtime_parameters(
            parameter_values=parameter_values)
        config_builder.update_pipeline_root(pipeline_root=pipeline_root)
        updated_runtime_config = config_builder.build()
        pipeline_dict['runtimeConfig'] = updated_runtime_config

    # Creating job creation request to get the final request URL
    pipeline_jobs_api_url = f'https://{region}-{_CAIPP_ENDPOINT_WITHOUT_REGION}/{_CAIPP_API_VERSION}/projects/{project_id}/locations/{region}/pipelineJobs'

    # Preparing the request body for the Cloud Function processing
    pipeline_name = pipeline_dict['pipelineSpec']['pipelineInfo']['name']
    full_pipeline_name = 'projects/{}/pipelineJobs/{}'.format(project_id, pipeline_name)
    pipeline_display_name = pipeline_dict.get('displayName')
    time_format_suffix = "-{{$.scheduledTime.strftime('%Y-%m-%d-%H-%M-%S')}}"
    if 'name' in pipeline_dict:
        pipeline_dict['name'] += time_format_suffix
    if 'displayName' in pipeline_dict:
        pipeline_dict['displayName'] += time_format_suffix
{'fraction_non_alphanumeric': 0.06955481500469965, 'fraction_numerical': 0.0037597197299837648, 'mean_word_length': 4.044827586206897, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 4, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '966833', 'n_tokens_mistral': 3261, 'n_tokens_neox': 3225, 'n_words': 1000}