<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.

# %%
import os
from typing import Dict, List, Optional

from kubernetes import client as k8s_client
import yaml

import kfp

REPO_ROOT = os.path.join('..', '..')
SAMPLES_CONFIG_PATH = os.path.join(REPO_ROOT, 'samples', 'test', 'config.yaml')
SAMPLES_CONFIG = None
with open(SAMPLES_CONFIG_PATH, 'r') as stream:
    SAMPLES_CONFIG = yaml.safe_load(stream)

download_gcs_tgz = kfp.components.load_component_from_file(
    'components/download_gcs_tgz.yaml')
run_sample = kfp.components.load_component_from_file(
    'components/run_sample.yaml')
kaniko = kfp.components.load_component_from_file('components/kaniko.yaml')
build_go = kfp.components.load_component_from_file('components/build_go.yaml')

PIPELINE_TIME_OUT = 40 * 60  # 40 minutes


@kfp.dsl.pipeline(name='v2 sample test')
def v2_sample_test(
    context: 'URI' = 'gs://your-bucket/path/to/context.tar.gz',
    gcs_root: 'URI' = 'gs://ml-pipeline-test/v2',
    image_registry: 'URI' = 'gcr.io/ml-pipeline-test',
    kfp_host: 'URI' = 'http://ml-pipeline:8888',
    samples_config: List[Dict] = SAMPLES_CONFIG,
    kfp_package_path:
    'URI' = 'git+https://github.com/kubeflow/pipelines#egg=kfp&subdirectory=sdk/python'
):
    # pipeline configs
    conf = kfp.dsl.get_pipeline_conf()
    conf.set_timeout(
        PIPELINE_TIME_OUT
    )  # add timeout to avoid pipelines stuck in running leak indefinetely

    download_src_op = download_gcs_tgz(gcs_path=context).set_cpu_limit(
        '0.5').set_memory_limit('500Mi').set_display_name('download_src')
    download_src_op.execution_options.caching_strategy.max_cache_staleness = "P0D"

    def build_image(name: str, dockerfile: str) -> kfp.dsl.ContainerOp:
        task: kfp.dsl.ContainerOp = kaniko(
            context_artifact=download_src_op.outputs['folder'],
            destination=f'{image_registry}/{name}',
            dockerfile=dockerfile,
        )
        # CPU request/limit can be more flexible (request < limit), because being assigned to a node
        # with insufficient CPU resource will only slow the task down, but not fail.
        task.container.set_cpu_request('1').set_cpu_limit('2')
        # Memory request/limit needs to be more rigid (request == limit), because in a node without
        # enough memory, the task can hang indefinetely or OOM.
        task.container.set_memory_request('4Gi').set_memory_limit('4Gi')
        task.set_display_name(f'build-image-{name}')
        task.set_retry(
            1, policy='Always'
        )  # Always -> retry on both system error and user code failure.
        return task

    # build v2 go images
    build_go_op = build_go(
        destination=f'{image_registry}/kfp-',
        context=download_src_op.outputs['folder'],
    )
    build_go_op.set_retry(1, policy='Always')
    build_go_op.container.set_cpu_request('1').set_cpu_limit('2')
    build_go_op.container.set_memory_request('4Gi').set_memory_limit('4Gi')

    # build sample test image
    build_samples_image_op = build_image(
        name='v2-sample-test',
        dockerfile='v2/test/Dockerfile',
    )
    # build v2 engine images
    build_kfp_launcher_v2_op = build_image(
        name='kfp-launcher-v2',
        dockerfile='v2/container/launcher-v2/Dockerfile')
    build_kfp_driver_op = build_image(
        name='kfp-driver', dockerfile='v2/container/driver/Dockerfile')

    # run test samples in parallel
    with kfp.dsl.ParallelFor(samples_config) as sample:
        run_sample_op: kfp.dsl.ContainerOp = run_sample(
            name=sample.name,
            sample_path=sample.path,
            gcs_root=gcs_root,
            external_host=kfp_host,
            launcher_image=build_go_op.outputs['digest_launcher'],
            launcher_v2_image=build_go_op.outputs['digest_launcher_v2'],
            driver_image=build_go_op.outputs['digest_driver'],
            backend_compiler=build_go_op.outputs['backend_compiler'],
        )
        run_sample_op.container.image = build_samples_image_op.outputs['digest']
        run_sample_op.set_display_name(f'sample_{sample.name}')
        run_sample_op.set_retry(1, policy='Always')

        run_sample_op.container.add_env_variable(<commit_msg>test: Remove old v2 launcher building steps. (#6247)<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.

# %%
import os
from typing import Dict, List, Optional

from kubernetes import client as k8s_client
import yaml

import kfp

REPO_ROOT = os.path.join('..', '..')
SAMPLES_CONFIG_PATH = os.path.join(REPO_ROOT, 'samples', 'test', 'config.yaml')
SAMPLES_CONFIG = None
with open(SAMPLES_CONFIG_PATH, 'r') as stream:
    SAMPLES_CONFIG = yaml.safe_load(stream)

download_gcs_tgz = kfp.components.load_component_from_file(
    'components/download_gcs_tgz.yaml')
run_sample = kfp.components.load_component_from_file(
    'components/run_sample.yaml')
kaniko = kfp.components.load_component_from_file('components/kaniko.yaml')
build_go = kfp.components.load_component_from_file('components/build_go.yaml')

PIPELINE_TIME_OUT = 40 * 60  # 40 minutes


@kfp.dsl.pipeline(name='v2 sample test')
def v2_sample_test(
    context: 'URI' = 'gs://your-bucket/path/to/context.tar.gz',
    gcs_root: 'URI' = 'gs://ml-pipeline-test/v2',
    image_registry: 'URI' = 'gcr.io/ml-pipeline-test',
    kfp_host: 'URI' = 'http://ml-pipeline:8888',
    samples_config: List[Dict] = SAMPLES_CONFIG,
    kfp_package_path:
    'URI' = 'git+https://github.com/kubeflow/pipelines#egg=kfp&subdirectory=sdk/python'
):
    # pipeline configs
    conf = kfp.dsl.get_pipeline_conf()
    conf.set_timeout(
        PIPELINE_TIME_OUT
    )  # add timeout to avoid pipelines stuck in running leak indefinetely

    download_src_op = download_gcs_tgz(gcs_path=context).set_cpu_limit(
        '0.5').set_memory_limit('500Mi').set_display_name('download_src')
    download_src_op.execution_options.caching_strategy.max_cache_staleness = "P0D"

    def build_image(name: str, dockerfile: str) -> kfp.dsl.ContainerOp:
        task: kfp.dsl.ContainerOp = kaniko(
            context_artifact=download_src_op.outputs['folder'],
            destination=f'{image_registry}/{name}',
            dockerfile=dockerfile,
        )
        # CPU request/limit can be more flexible (request < limit), because being assigned to a node
        # with insufficient CPU resource will only slow the task down, but not fail.
        task.container.set_cpu_request('1').set_cpu_limit('2')
        # Memory request/limit needs to be more rigid (request == limit), because in a node without
        # enough memory, the task can hang indefinetely or OOM.
        task.container.set_memory_request('4Gi').set_memory_limit('4Gi')
        task.set_display_name(f'build-image-{name}')
        task.set_retry(
            1, policy='Always'
        )  # Always -> retry on both system error and user code failure.
        return task

    # build v2 go images
    build_go_op = build_go(
        destination=f'{image_registry}/kfp-',
        context=download_src_op.outputs['folder'],
    )
    build_go_op.set_retry(1, policy='Always')
    build_go_op.container.set_cpu_request('1').set_cpu_limit('2')
    build_go_op.container.set_memory_request('4Gi').set_memory_limit('4Gi')

    # build sample test image
    build_samples_image_op = build_image(
        name='v2-sample-test',
        dockerfile='v2/test/Dockerfile',
    )

    # run test samples in parallel
    with kfp.dsl.ParallelFor(samples_config) as sample:
        run_sample_op: kfp.dsl.ContainerOp = run_sample(
            name=sample.name,
            sample_path=sample.path,
            gcs_root=gcs_root,
            external_host=kfp_host,
            launcher_image=build_go_op.outputs['digest_launcher'],
            launcher_v2_image=build_go_op.outputs['digest_launcher_v2'],
            driver_image=build_go_op.outputs['digest_driver'],
            backend_compiler=build_go_op.outputs['backend_compiler'],
        )
        run_sample_op.container.image = build_samples_image_op.outputs['digest']
        run_sample_op.set_display_name(f'sample_{sample.name}')
        run_sample_op.set_retry(1, policy='Always')

        run_sample_op.container.add_env_variable(
{'fraction_non_alphanumeric': 0.10748519116855143, 'fraction_numerical': 0.010662358642972537, 'mean_word_length': 3.6476476476476476, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7184401', 'n_tokens_mistral': 3069, 'n_tokens_neox': 2978, 'n_words': 624}