Update tests and CLI to new Tensorlake SDK

Author: eabatalov

indexify/src/indexify/cli/__init__.py

@@ -1,18 +1,18 @@
 import click
 
-from . import build_image, deploy, executor
+from . import build_images, deploy, executor
 
 
 @click.group()
 @click.version_option(package_name="indexify", prog_name="indexify-cli")
 @click.pass_context
 def cli(ctx: click.Context):
     """
-    Indexify CLI to manage and deploy workflows to Indexify Server and run Indexify Executors.
+    Indexify CLI to manage and deploy applications to Indexify Server and run Indexify Executors.
     """
     pass
 
 
-cli.add_command(build_image.build_image)
+cli.add_command(build_images.build_images)
 cli.add_command(deploy.deploy)
 cli.add_command(executor.executor)

indexify/src/indexify/cli/build_image.py renamed to indexify/src/indexify/cli/build_images.py

@@ -1,46 +1,43 @@
 import importlib
-from typing import Any, Generator, Tuple
+import os
+from typing import Any, Generator, Set
 
 import click
 import docker
 import docker.api.build
 from docker.errors import BuildError
-from tensorlake.functions_sdk.image import Image
-from tensorlake.functions_sdk.workflow_module import (
-    WorkflowModuleInfo,
-    load_workflow_module_info,
+from docker.models.images import Image as DockerImage
+from tensorlake.applications import Image
+from tensorlake.applications.image import (
+    ImageInformation,
+    dockerfile_content,
+    image_infos,
 )
+from tensorlake.applications.remote.application.ignored_code_paths import (
+    ignored_code_paths,
+)
+from tensorlake.applications.remote.application.loader import load_application
 
 
 @click.command(
-    short_help="Build images for graphs/workflows defined in the workflow file"
+    short_help="Builds images for application defined in <application-source-path> directory or file"
 )
 @click.argument(
-    "workflow-file-path",
-    type=click.Path(exists=True, file_okay=True, dir_okay=False),
-)
-@click.option(
-    "-i",
-    "--image-names",
-    multiple=True,
-    help="Names of images to build. Can be specified multiple times. If not provided, all images will be built.",
+    "application-path",
+    type=click.Path(exists=True, file_okay=True, dir_okay=True),
 )
-def build_image(
-    workflow_file_path: str,
-    image_names: tuple[str, ...] = None,
-):
-    """
-    Build the images associated to an Indexify workflow
-
-    A workflow is defined in a Python file, and the images are built using the local Docker daemon.
-    """
+def build_images(application_path: str):
     try:
-        workflow_module_info: WorkflowModuleInfo = load_workflow_module_info(
-            workflow_file_path
+        application_source_dir_or_file_path: str = os.path.abspath(application_path)
+
+        ignored_absolute_paths: Set[str] = ignored_code_paths(
+            os.path.dirname(application_source_dir_or_file_path)
         )
+
+        load_application(application_source_dir_or_file_path, ignored_absolute_paths)
     except Exception as e:
         click.secho(
-            f"Failed loading workflow file, please check the error message: {e}",
+            f"Failed to load the application modules, please check the error message: {e}",
             fg="red",
         )
         raise click.Abort
@@ -49,39 +46,28 @@ def build_image(
     docker_client.ping()
 
     indexify_version: str = importlib.metadata.version("indexify")
-    for image in workflow_module_info.images.keys():
+    for image, image_info in image_infos().items():
         image: Image
-        if len(image_names) > 0 and image.image_name not in image_names:
-            click.echo(
-                f"Skipping image `{image.image_name}` as it is not in the provided image names."
-            )
-            continue
-
-        click.echo(f"Building image `{image.image_name}`")
-
+        image_info: ImageInformation
+        click.echo(f"Building image `{image.name}:{image.tag}`")
         image.run(f"pip install 'indexify=={indexify_version}'")
-        built_image, logs_generator = _build(image=image, docker_client=docker_client)
+
         try:
             built_image, logs_generator = _build(
                 image=image, docker_client=docker_client
             )
+            built_image: DockerImage
             _print_build_log(logs_generator)
-            click.secho(f"built image: {built_image.tags[0]}", fg="green")
+            click.secho(f"Built image: {built_image.tags[0]}", fg="green")
         except BuildError as e:
             raise click.Abort() from e
 
-        click.secho(f"built image: {built_image.tags[0]}", fg="green")
-
 
 def _build(
     image: Image, docker_client: docker.DockerClient
-) -> Tuple[docker.models.images.Image, Generator[str, Any, None]]:
-    docker_file = image.dockerfile()
-    image_name = (
-        image.image_name
-        if ":" in image.image_name
-        else f"{image.image_name}:{image.image_tag}"
-    )
+) -> tuple[DockerImage, Generator[str, Any, None]]:
+    docker_file_content: str = dockerfile_content(image)
+    image_name = image.name if ":" in image.name else f"{image.name}:{image.tag}"
 
     docker.api.build.process_dockerfile = lambda dockerfile, path: (
         "Dockerfile",
@@ -91,7 +77,7 @@ def _build(
     try:
         built_image, logs_generator = docker_client.images.build(
             path=".",
-            dockerfile=docker_file,
+            dockerfile=docker_file_content,
             tag=image_name,
             rm=True,
             # pull=True,  # optional: ensures fresh base images

indexify/src/indexify/cli/deploy.py

@@ -1,57 +1,41 @@
+import traceback
+
 import click
-from tensorlake import Graph
-from tensorlake.functions_sdk.graph_serialization import graph_code_dir_path
-from tensorlake.functions_sdk.remote_graph import RemoteGraph
-from tensorlake.functions_sdk.workflow_module import (
-    WorkflowModuleInfo,
-    load_workflow_module_info,
-)
+from tensorlake.applications.remote.deploy import deploy as tl_deploy
 
 
 @click.command(
-    short_help="Deploy all graphs/workflows defined in the workflow file to Indexify"
+    short_help="Deploys application defined in <application-path> directory or file to Indexify"
 )
-# Path to the file where the graphs/workflows are defined as global variables
 @click.argument(
-    "workflow-file-path",
-    type=click.Path(exists=True, file_okay=True, dir_okay=False),
+    "application-path",
+    type=click.Path(exists=True, file_okay=True, dir_okay=True),
 )
 @click.option(
     "-u",
-    "--upgrade-queued-invocations",
+    "--upgrade-running-requests",
     is_flag=True,
     default=False,
-    help="Upgrade invocations that are already queued or running to use the deployed version of the graphs/workflows",
+    help="Upgrade requests that are already queued or running to use the new deployed version of the application",
 )
 def deploy(
-    workflow_file_path: str,
-    upgrade_queued_invocations: bool,
+    application_path: str,
+    upgrade_running_requests: bool,
 ):
-    click.echo(f"Preparing deployment for {workflow_file_path}")
+    click.echo(f"Preparing deployment for application from {application_path}")
+
     try:
-        workflow_module_info: WorkflowModuleInfo = load_workflow_module_info(
-            workflow_file_path
+        tl_deploy(
+            application_source_dir_or_file_path=application_path,
+            upgrade_running_requests=upgrade_running_requests,
+            load_application_modules=True,
         )
     except Exception as e:
         click.secho(
-            f"Failed loading workflow file, please check the error message: {e}",
+            f"Application could not be deployed, please check the error message:",
             fg="red",
         )
+        traceback.print_exception(e)
         raise click.Abort
 
-    for graph in workflow_module_info.graphs:
-        graph: Graph
-        try:
-            RemoteGraph.deploy(
-                graph,
-                code_dir_path=graph_code_dir_path(workflow_file_path),
-                upgrade_tasks_to_latest_version=upgrade_queued_invocations,
-            )
-        except Exception as e:
-            click.secho(
-                f"Graph {graph.name} could not be deployed, please check the error message: {e}",
-                fg="red",
-            )
-            raise click.Abort
-
-        click.secho(f"Deployed {graph.name}", fg="green")
+    click.secho(f"Successfully deployed the application", fg="green")

indexify/src/indexify/cli/executor.py

@@ -10,7 +10,7 @@
 from importlib.metadata import version
 from pathlib import Path
 from socket import gethostname
-from typing import Dict, List, Optional
+from typing import Dict, List
 
 import click
 import nanoid
@@ -110,7 +110,7 @@ def executor(
     verbose: bool,
     very_verbose: bool,
     function_uris: List[str],
-    config_path: Optional[str],
+    config_path: str | None,
     executor_cache_path: str,
     monitoring_server_host: str,
     monitoring_server_port: int,

indexify/src/indexify/proto/executor_api.proto

@@ -293,6 +293,7 @@ message ReduceOp {
     // The DataPayload of the last function call in the reducer will have the function call id set
     // as the id of the ReduceOp.
   optional string id = 1;
+  // Contains at least two items.
   repeated FunctionArg  collection = 2; 
   optional FunctionRef reducer     = 3;
   // This required metadata allows SDK to restore original function call from the arguments.

indexify/tests/cli/test_environment_variables.py

@@ -4,8 +4,8 @@
 from typing import Dict
 
 import pydantic
-import tensorlake.workflows.interface as tensorlake
-from tensorlake.workflows.remote.deploy import deploy
+from tensorlake.applications import Request, api, call_remote_api, function
+from tensorlake.applications.remote.deploy import deploy
 from testing import (
     ExecutorProcessContextManager,
     executor_pid,
@@ -18,8 +18,8 @@ class Response(pydantic.BaseModel):
     environment: Dict[str, str]
 
 
-@tensorlake.api()
-@tensorlake.function()
+@api()
+@function()
 def function_a(_a: int) -> Response:
     return Response(executor_pid=executor_pid(), environment=os.environ.copy())
 
@@ -45,7 +45,7 @@ def test_executor_env_variables_are_passed_to_functions(self):
             wait_executor_startup(7001)
 
             for _ in range(10):
-                request: tensorlake.Request = tensorlake.call_remote_api(
+                request: Request = call_remote_api(
                     function_a,
                     1,
                 )

indexify/tests/cli/test_executor_exit.py

@@ -3,17 +3,17 @@
 import unittest
 from typing import List
 
-import tensorlake.workflows.interface as tensorlake
-from tensorlake.workflows.remote.deploy import deploy
+from tensorlake.applications import Request, api, call_remote_api, function
+from tensorlake.applications.remote.deploy import deploy
 from testing import (
     ExecutorProcessContextManager,
     function_uri,
     wait_executor_startup,
 )
 
 
-@tensorlake.api()
-@tensorlake.function()
+@api()
+@function()
 def success_func(sleep_secs: float) -> str:
     time.sleep(sleep_secs)
     return "success"
@@ -39,10 +39,10 @@ def test_all_tasks_succeed_when_executor_exits(self):
             print(f"Started Executor A with PID: {executor_a.pid}")
             wait_executor_startup(7001)
 
-            requests: List[tensorlake.Request] = []
+            requests: List[Request] = []
             for i in range(10):
                 print(f"Running request {i}")
-                request: tensorlake.Request = tensorlake.call_remote_api(
+                request: Request = call_remote_api(
                     success_func,
                     0.1,
                 )

indexify/tests/cli/test_metrics.py

@@ -5,13 +5,13 @@
 from typing import Dict
 
 import httpx
-import tensorlake.workflows.interface as tensorlake
 
 # We're using internal APIs here, this might break when we update prometheus_client.
 from prometheus_client.metrics_core import Metric
 from prometheus_client.parser import text_string_to_metric_families
 from prometheus_client.samples import Sample
-from tensorlake.workflows.remote.deploy import deploy
+from tensorlake.applications import Request, api, call_remote_api, function
+from tensorlake.applications.remote.deploy import deploy
 from testing import (
     ExecutorProcessContextManager,
     wait_executor_startup,
@@ -29,8 +29,8 @@ def fetch_metrics(
     return metrics
 
 
-@tensorlake.api()
-@tensorlake.function()
+@api()
+@function()
 def successful_function(arg: str) -> str:
     return "success"
 
@@ -68,7 +68,7 @@ def test_executor_info(self):
             self.assertIn("id", info_sample.labels)
 
     def test_expected_function_executor_infos(self):
-        request: tensorlake.Request = tensorlake.call_remote_api(
+        request: Request = call_remote_api(
             successful_function,
             "ignored",
         )