Improved config system in flytekit and improved FlyteRemote

.github/workflows/pythonbuild.yml

@@ -6,6 +6,9 @@ on:
     - master
   pull_request:
 
+env:
+  FLYTE_SDK_LOGGING_LEVEL: 10  # debug
+
 jobs:
   build:
     runs-on: ubuntu-latest

.gitignore

@@ -28,3 +28,4 @@ _build/
 docs/source/generated/
 .pytest_flyte
 htmlcov
+*.ipynb

flytekit/__init__.py

@@ -155,12 +155,15 @@
 
 import sys
 
+import flytekit.configuration.internal
+
 if sys.version_info < (3, 10):
     from importlib_metadata import entry_points
 else:
     from importlib.metadata import entry_points
 
-from flytekit.configuration.sdk import USE_STRUCTURED_DATASET
+from flytekit import configuration
+from flytekit.configuration import internal as _internal
 from flytekit.core.base_sql_task import SQLTask
 from flytekit.core.base_task import SecurityContext, TaskMetadata, kwtypes
 from flytekit.core.checkpointer import Checkpoint
@@ -190,7 +193,7 @@
 from flytekit.models.types import LiteralType
 from flytekit.types import directory, file, schema
 
-if USE_STRUCTURED_DATASET.get():
+if _internal.LocalSDK.USE_STRUCTURED_DATASET.read():
     from flytekit.types.structured.structured_dataset import (
         StructuredDataset,
         StructuredDatasetFormat,

flytekit/clis/helpers.py

@@ -6,7 +6,7 @@
 from flyteidl.core import identifier_pb2 as _identifier_pb2
 from flyteidl.core import workflow_pb2 as _workflow_pb2
 
-from flytekit.clis.sdk_in_container.serialize import _DOMAIN_PLACEHOLDER, _PROJECT_PLACEHOLDER, _VERSION_PLACEHOLDER
+from flytekit.configuration import DOMAIN_PLACEHOLDER, PROJECT_PLACEHOLDER, VERSION_PLACEHOLDER
 
 
 def parse_args_into_dict(input_arguments):
@@ -33,13 +33,13 @@ def str2bool(str):
 def _hydrate_identifier(
     project: str, domain: str, version: str, identifier: _identifier_pb2.Identifier
 ) -> _identifier_pb2.Identifier:
-    if not identifier.project or identifier.project == _PROJECT_PLACEHOLDER:
+    if not identifier.project or identifier.project == PROJECT_PLACEHOLDER:
         identifier.project = project
 
-    if not identifier.domain or identifier.domain == _DOMAIN_PLACEHOLDER:
+    if not identifier.domain or identifier.domain == DOMAIN_PLACEHOLDER:
         identifier.domain = domain
 
-    if not identifier.version or identifier.version == _VERSION_PLACEHOLDER:
+    if not identifier.version or identifier.version == VERSION_PLACEHOLDER:
         identifier.version = version
     return identifier
 

flytekit/clis/sdk_in_container/constants.py

@@ -6,6 +6,7 @@
 CTX_TEST = "test"
 CTX_PACKAGES = "pkgs"
 CTX_NOTIFICATIONS = "notifications"
+CTX_CONFIG_FILE = "config_file"
 
 
 project_option = _click.option(

flytekit/clis/sdk_in_container/package.py

@@ -2,46 +2,18 @@
 import sys
 import tarfile
 import tempfile
-import typing
 
 import click
 
-from flytekit.clis.sdk_in_container import constants, serialize
-from flytekit.configuration import internal
+from flytekit.clis.sdk_in_container import constants
+from flytekit.configuration import (
+    DEFAULT_RUNTIME_PYTHON_INTERPRETER,
+    FastSerializationSettings,
+    ImageConfig,
+    SerializationSettings,
+)
 from flytekit.core import context_manager
-from flytekit.core.context_manager import ImageConfig, look_up_image_info
-from flytekit.tools import fast_registration, module_loader
-
-_DEFAULT_IMAGE_NAME = "default"
-_DEFAULT_RUNTIME_PYTHON_INTERPRETER = "/opt/venv/bin/python3"
-
-
-def validate_image(ctx: typing.Any, param: str, values: tuple) -> ImageConfig:
-    """
-    Validates the image to match the standard format. Also validates that only one default image
-    is provided. a default image, is one that is specified as
-      default=img or just img. All other images should be provided with a name, in the format
-      name=img
-    """
-    default_image = None
-    images = []
-    for v in values:
-        if "=" in v:
-            splits = v.split("=", maxsplit=1)
-            img = look_up_image_info(name=splits[0], tag=splits[1], optional_tag=False)
-        else:
-            img = look_up_image_info(_DEFAULT_IMAGE_NAME, v, False)
-
-        if default_image and img.name == _DEFAULT_IMAGE_NAME:
-            raise click.BadParameter(
-                f"Only one default image can be specified. Received multiple {default_image} & {img} for {param}"
-            )
-        if img.name == _DEFAULT_IMAGE_NAME:
-            default_image = img
-        else:
-            images.append(img)
-
-    return ImageConfig(default_image, images)
+from flytekit.tools import fast_registration, module_loader, serialize_helpers
 
 
 @click.command("package")
@@ -52,7 +24,7 @@ def validate_image(ctx: typing.Any, param: str, values: tuple) -> ImageConfig:
     required=False,
     multiple=True,
     type=click.UNPROCESSED,
-    callback=validate_image,
+    callback=ImageConfig.validate_image,
     help="A fully qualified tag for an docker image, e.g. somedocker.com/myimage:someversion123. This is a "
     "multi-option and can be of the form --image xyz.io/docker:latest"
     " --image my_image=xyz.io/docker2:latest. Note, the `name=image_uri`. The name is optional, if not"
@@ -95,7 +67,7 @@ def validate_image(ctx: typing.Any, param: str, values: tuple) -> ImageConfig:
 @click.option(
     "-p",
     "--python-interpreter",
-    default=_DEFAULT_RUNTIME_PYTHON_INTERPRETER,
+    default=DEFAULT_RUNTIME_PYTHON_INTERPRETER,
     required=False,
     help="Use this to override the default location of the in-container python interpreter that will be used by "
     "Flyte to load your program. This is usually where you install flytekit within the container.",
@@ -120,23 +92,13 @@ def package(ctx, image_config, source, output, force, fast, in_container_source_
     if os.path.exists(output) and not force:
         raise click.BadParameter(click.style(f"Output file {output} already exists, specify -f to override.", fg="red"))
 
-    env_binary_path = os.path.dirname(python_interpreter)
-    venv_root = os.path.dirname(env_binary_path)
-    serialization_settings = context_manager.SerializationSettings(
-        project=serialize._PROJECT_PLACEHOLDER,
-        domain=serialize._DOMAIN_PLACEHOLDER,
-        version=serialize._VERSION_PLACEHOLDER,
-        fast_serialization_settings=context_manager.FastSerializationSettings(
+    serialization_settings = SerializationSettings(
+        image_config=image_config,
+        fast_serialization_settings=FastSerializationSettings(
             enabled=fast,
             destination_dir=in_container_source_path,
         ),
-        image_config=image_config,
-        env={internal.IMAGE.env_var: image_config.default_image.full},  # TODO this env variable should be deprecated
-        flytekit_virtualenv_root=venv_root,
         python_interpreter=python_interpreter,
-        entrypoint_settings=context_manager.EntrypointSettings(
-            path=os.path.join(venv_root, serialize._DEFAULT_FLYTEKIT_RELATIVE_ENTRYPOINT_LOC)
-        ),
     )
 
     pkgs = ctx.obj[constants.CTX_PACKAGES]
@@ -151,11 +113,11 @@ def package(ctx, image_config, source, output, force, fast, in_container_source_
             click.secho(f"Loading packages {pkgs} under source root {source}", fg="yellow")
             module_loader.just_load_modules(pkgs=pkgs)
 
-        registrable_entities = serialize.get_registrable_entities(ctx)
+        registrable_entities = serialize_helpers.get_registrable_entities(ctx)
 
         if registrable_entities:
             with tempfile.TemporaryDirectory() as output_tmpdir:
-                serialize.persist_registrable_entities(registrable_entities, output_tmpdir)
+                serialize_helpers.persist_registrable_entities(registrable_entities, output_tmpdir)
 
                 # If Fast serialization is enabled, then an archive is also created and packaged
                 if fast:

flytekit/clis/sdk_in_container/pyflyte.py

@@ -1,18 +1,12 @@
-import os as _os
-from pathlib import Path
-
 import click
 
-from flytekit.clis.sdk_in_container.constants import CTX_PACKAGES
+from flytekit import configuration
+from flytekit.clis.sdk_in_container.constants import CTX_CONFIG_FILE, CTX_PACKAGES
 from flytekit.clis.sdk_in_container.init import init
 from flytekit.clis.sdk_in_container.local_cache import local_cache
 from flytekit.clis.sdk_in_container.package import package
 from flytekit.clis.sdk_in_container.serialize import serialize
-from flytekit.configuration import platform as _platform_config
-from flytekit.configuration import set_flyte_config_file
-from flytekit.configuration.internal import CONFIGURATION_PATH
-from flytekit.configuration.platform import URL as _URL
-from flytekit.configuration.sdk import WORKFLOW_PACKAGES as _WORKFLOW_PACKAGES
+from flytekit.configuration.internal import LocalSDK
 
 
 def validate_package(ctx, param, values):
@@ -25,13 +19,6 @@ def validate_package(ctx, param, values):
 
 
 @click.group("pyflyte", invoke_without_command=True)
-@click.option(
-    "-c",
-    "--config",
-    required=False,
-    type=str,
-    help="Path to config file for use within container",
-)
 @click.option(
     "-k",
     "--pkgs",
@@ -42,60 +29,31 @@ def validate_package(ctx, param, values):
     "option will override the option specified in the configuration file, or environment variable",
 )
 @click.option(
-    "-i",
-    "--insecure",
+    "-c",
+    "--config",
     required=False,
-    type=bool,
-    help="Disable SSL when connecting to Flyte backend.",
+    type=str,
+    help="Path to config file for use within container",
 )
 @click.pass_context
-def main(ctx, config=None, pkgs=None, insecure=None):
+def main(ctx, pkgs=None, config=None):
     """
     Entrypoint for all the user commands.
     """
-    update_configuration_file(config)
-
     ctx.obj = dict()
 
-    # Determine SSL.  Note that the insecure option in this command is not a flag because we want to look
-    # up configuration settings if it's missing.  If the command line option says insecure but the config object
-    # says no, let's override the config object by overriding the environment variable.
-    if insecure and not _platform_config.INSECURE.get():
-        _platform_config.INSECURE.get()
-        _os.environ[_platform_config.INSECURE.env_var] = "True"
-
     # Handle package management - get from config if not specified on the command line
     pkgs = pkgs or []
-    if len(pkgs) == 0:
-        pkgs = _WORKFLOW_PACKAGES.get()
+    if config:
+        ctx.obj[CTX_CONFIG_FILE] = config
+        cfg = configuration.ConfigFile(config)
+        if not pkgs:
+            pkgs = LocalSDK.WORKFLOW_PACKAGES.read(cfg)
+            if pkgs is None:
+                pkgs = []
     ctx.obj[CTX_PACKAGES] = pkgs
 
 
-def update_configuration_file(config_file_path):
-    """
-    Changes the configuration singleton object to read from another file if specified, which should be
-    at the base of the repository.
-
-    :param Text config_file_path:
-    """
-    configuration_file = Path(config_file_path or CONFIGURATION_PATH.get())
-    if configuration_file.is_file():
-        click.secho(
-            "Using configuration file at {}".format(configuration_file.absolute().as_posix()),
-            fg="green",
-        )
-        set_flyte_config_file(configuration_file.as_posix())
-    else:
-        click.secho(
-            "Configuration file '{}' could not be loaded. Using values from environment.".format(
-                CONFIGURATION_PATH.get()
-            ),
-            color="yellow",
-        )
-        set_flyte_config_file(None)
-    click.secho("Flyte Admin URL {}".format(_URL.get()), fg="green")
-
-
 main.add_command(serialize)
 main.add_command(package)
 main.add_command(local_cache)