Improve components tests (#208)

Summary:
Pull Request resolved: #208

The diff introduces the following changes:
* Improved final test report
* Make sure that all resources allocated during the run are deleted
* Fail the test if any of the components fail
* Reduce timeout to 10 minutes.
* Make sure that there is a single logic that governs the execution mode: `dryrun` or normal

Reviewed By: d4l3k

Differential Revision: D31247636

fbshipit-source-id: ce587c88a9309f83c72d9aaae839f7945fe3b5a2

Author: aivanou

.github/workflows/components-integration-tests.yaml

@@ -43,11 +43,6 @@ jobs:
         env:
           AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
           AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
-        run: |
-          if [ -z "$AWS_ACCESS_KEY_ID" ]; then
-            # only dryrun if no secrets
-            ARGS="--dryrun"
-          else
-            ARGS=
-          fi
-          scripts/component_integration_tests.py $ARGS
+          INTEGRATION_TEST_STORAGE: ${{ secrets.INTEGRATION_TEST_STORAGE }}
+          CONTAINER_REPO: ${{ secrets.CONTAINER_REPO }}
+        run: scripts/component_integration_tests.py

scripts/component_integration_tests.py

@@ -8,7 +8,6 @@
 """
 Kubernetes integration tests.
 """
-import argparse
 import os
 
 # pyre-ignore-all-errors[21] # Cannot find module utils
@@ -51,31 +50,26 @@ def get_local_docker_sched_info(image: str) -> SchedulerInfo:
 
 
 def main() -> None:
-    parser = argparse.ArgumentParser(description="kubernetes integration test runner")
-    parser.add_argument(
-        "--dryrun",
-        action="store_true",
-        help="Does not actually submit the app," " just prints the scheduler request",
-    )
-    args = parser.parse_args()
     print("Starting components integration tests")
     torchx_image = "dummy_image"
     examples_image = "dummy_image"
+    dryrun: bool = False
     try:
         build = build_and_push_image()
         torchx_image = build.torchx_image
         examples_image = build.examples_image
     except MissingEnvError:
+        dryrun = True
         print("Skip runnig tests, executed only docker buid step")
-    test_suite = IntegComponentTest(timeout=900)  # 15 minutes
+    test_suite = IntegComponentTest(timeout=600)  # 10 minutes
     test_suite.run_components(
         component_provider,
         scheduler_infos=[
             get_local_cwd_sched_info(os.getcwd()),
             get_local_docker_sched_info(torchx_image),
             get_k8s_sched_info(torchx_image),
         ],
-        dryrun=args.dryrun,
+        dryrun=dryrun,
     )
 
     test_suite.run_components(
@@ -84,7 +78,7 @@ def main() -> None:
             get_local_docker_sched_info(examples_image),
             get_k8s_sched_info(examples_image),
         ],
-        dryrun=args.dryrun,
+        dryrun=dryrun,
     )
 
 

torchx/components/integration_tests/component_provider.py

@@ -5,6 +5,7 @@
 # LICENSE file in the root directory of this source tree.
 
 import os
+import tempfile
 from abc import ABC, abstractmethod
 
 import torchx.components.dist as dist_components
@@ -35,7 +36,7 @@ def tearDown(self) -> None:
 
 class DDPComponentProvider(ComponentProvider):
     def get_app_def(self) -> AppDef:
-        args = ["torchx.apps.test.dummy_script"]
+        args = ["torchx.components.integration_tests.test.dummy_app"]
         rdzv_backend: str = "c10d"
         rdzv_endpoint: str = "localhost:29400"
         return dist_components.ddp(
@@ -81,19 +82,20 @@ def __init__(self, image: str, scheduler: SchedulerBackend) -> None:
         self._file_path = "<None>"
 
     def setUp(self) -> None:
-        if self._scheduler == "local":
-            self._file_path: str = os.path.join(os.getcwd(), "test.txt")
+        fname = "torchx_touch_test.txt"
+        if self._scheduler == "local_cwd":
+            self._file_path: str = os.path.join(tempfile.gettempdir(), fname)
         else:
-            self._file_path: str = "test.txt"
+            self._file_path: str = fname
 
     def get_app_def(self) -> AppDef:
         return utils_components.touch(
-            file="test.txt",
+            file=self._file_path,
             image=self._image,
         )
 
     def tearDown(self) -> None:
-        if self._scheduler == "local" and os.path.exists(self._file_path):
+        if os.path.exists(self._file_path):
             os.remove(self._file_path)
 
 
@@ -104,9 +106,10 @@ def __init__(self, image: str, scheduler: SchedulerBackend) -> None:
         self._dst_path = "<None>"
 
     def setUp(self) -> None:
-        if self._scheduler == "local":
-            self._src_path: str = os.path.join(os.getcwd(), "copy_test.txt")
-            self._dst_path: str = os.path.join(os.getcwd(), "copy_test.txt.copy")
+        if self._scheduler == "local_cwd":
+            fname = "torchx_copy_test.txt"
+            self._src_path: str = os.path.join(tempfile.gettempdir(), fname)
+            self._dst_path: str = os.path.join(tempfile.gettempdir(), f"{fname}.copy")
             self._process_local_sched()
         else:
             self._src_path: str = "README.md"
@@ -118,7 +121,9 @@ def _process_local_sched(self) -> None:
                 f.write("test data")
 
     def tearDown(self) -> None:
-        if self._scheduler == "local" and os.path.exists(self._dst_path):
+        if os.path.exists(self._dst_path):
+            os.remove(self._dst_path)
+        if self._scheduler == "local_cwd" and os.path.exists(self._dst_path):
             os.remove(self._dst_path)
 
     def get_app_def(self) -> AppDef:

torchx/components/integration_tests/integ_tests.py

@@ -9,13 +9,14 @@
 import traceback
 from concurrent.futures import Future, ThreadPoolExecutor
 from dataclasses import dataclass
+from string import Template
 from types import ModuleType
-from typing import List, Dict, Optional, Tuple, Type, Callable, cast
+from typing import List, Optional, Tuple, Type, Callable, cast
 
 from pyre_extensions import none_throws
 from torchx.components.component_test_base import ComponentUtils
 from torchx.components.integration_tests.component_provider import ComponentProvider
-from torchx.specs import RunConfig, SchedulerBackend
+from torchx.specs import RunConfig, SchedulerBackend, AppDef
 
 
 @dataclass
@@ -27,13 +28,34 @@ class SchedulerInfo:
 
 @dataclass
 class AppDefRun:
-    provider_cls: Type[ComponentProvider]
+    provider: ComponentProvider
     scheduler_info: SchedulerInfo
+    app_def: AppDef
     future: Future
 
-    def __str__(self) -> str:
-        msg = f"`{self.scheduler_info.name}`:`{self.scheduler_info.image}`"
-        return msg
+
+_SUCCESS_APP_FORMAT_TEMPLATE = """
+  Name: ${name}
+  Scheduler: ${scheduler}"""
+
+_FAIL_APP_FORMAT_TEMPLATE = """
+  Name: ${name}
+  Provider: ${provider}
+  Scheduler: ${scheduler}
+  Image: ${image}
+  Error: ${error}"""
+
+_REPORT_FORMAT_TEMPLATE = """
+${boarder}
+Status: Success
+${boarder}
+${success_report}
+\n
+${boarder}
+Status: Failed
+${boarder}
+${fail_report}
+"""
 
 
 class IntegComponentTest:
@@ -61,14 +83,14 @@ def run_components(
         dryrun: bool = False,
     ) -> None:
         component_providers_cls = self._get_component_providers(module)
-        futures: List[AppDefRun] = []
+        app_runs: List[AppDefRun] = []
         executor = ThreadPoolExecutor()
         for scheduler_info in scheduler_infos:
             sched_futures = self._run_component_providers(
                 executor, component_providers_cls, scheduler_info, dryrun
             )
-            futures += sched_futures
-        self._wait_and_print_report(futures)
+            app_runs += sched_futures
+        self._wait_and_print_report(app_runs)
 
     def _get_component_providers(
         self, module: ModuleType
@@ -91,70 +113,94 @@ def _run_component_providers(
         scheduler_info: SchedulerInfo,
         dryrun: bool = False,
     ) -> List[AppDefRun]:
-        futures: List[AppDefRun] = []
+        app_runs: List[AppDefRun] = []
         for provider_cls in component_providers_cls:
+            provider = self._get_app_def_provider(provider_cls, scheduler_info)
             future = executor.submit(
-                self._run_component_provider, provider_cls, scheduler_info, dryrun
+                self._run_component_provider, provider, scheduler_info, dryrun
             )
-            futures.append(
+            app_runs.append(
                 AppDefRun(
-                    provider_cls=provider_cls,
+                    provider=provider,
                     scheduler_info=scheduler_info,
                     future=future,
+                    app_def=provider.get_app_def(),
                 )
             )
-        return futures
+        return app_runs
 
-    def _run_component_provider(
+    def _get_app_def_provider(
         self,
         component_provider_cls: Type[ComponentProvider],
         scheduler_info: SchedulerInfo,
+    ) -> ComponentProvider:
+        provider_cls = cast(Callable[..., ComponentProvider], component_provider_cls)
+        return none_throws(provider_cls(scheduler_info.name, scheduler_info.image))
+
+    def _run_component_provider(
+        self,
+        provider: ComponentProvider,
+        scheduler_info: SchedulerInfo,
         dryrun: bool = False,
-    ) -> str:
-        provider: Optional[ComponentProvider] = None
+    ) -> None:
         try:
-            provider_cls = cast(
-                Callable[..., ComponentProvider], component_provider_cls
-            )
-            provider = none_throws(
-                provider_cls(scheduler_info.name, scheduler_info.image)
-            )
             provider.setUp()
-            app_def = provider.get_app_def()
             ComponentUtils.run_appdef_on_scheduler(
-                app_def, scheduler_info.name, scheduler_info.runconfig, dryrun
+                provider.get_app_def(),
+                scheduler_info.name,
+                scheduler_info.runconfig,
+                dryrun,
             )
-            return app_def.name
         finally:
-            if provider:
-                provider.tearDown()
+            provider.tearDown()
 
-    def _wait_and_print_report(self, futures: List[AppDefRun]) -> None:
-        app_runs: Dict[str, List[str]] = {}
+    def _wait_and_print_report(self, app_runs: List[AppDefRun]) -> None:
+        succeeded_apps: List[AppDefRun] = []
+        failed_apps: List[Tuple[AppDefRun, str]] = []
         deadline = time.monotonic() + self._timeout
-        for future in futures:
+        for app_run in app_runs:
             task_timeout = max(0, int(deadline - time.monotonic()))
-            status, msg = self._get_app_def_run_status(future, task_timeout)
-            if status not in app_runs:
-                app_runs[status] = []
-            app_runs[status].append(msg)
-        for status, msgs in app_runs.items():
-            print(f"\nStatus: `{status}`\n")
-            print("\n".join(msgs))
-            print("\n")
+            error_msg = self._get_app_def_run_status(app_run, task_timeout)
+            if not error_msg:
+                succeeded_apps.append(app_run)
+            else:
+                failed_apps.append((app_run, error_msg))
+        success_report = ""
+        for app_run in succeeded_apps:
+            success_report_run = Template(_SUCCESS_APP_FORMAT_TEMPLATE).substitute(
+                name=app_run.app_def.name, scheduler=app_run.scheduler_info.name
+            )
+            success_report += f"{success_report_run}\n"
+        fail_report = ""
+        for app_run, error_msg in failed_apps:
+            fail_report_run = Template(_FAIL_APP_FORMAT_TEMPLATE).substitute(
+                name=app_run.app_def.name,
+                provider=app_run.provider,
+                scheduler=app_run.scheduler_info.name,
+                image=app_run.scheduler_info.image,
+                error=error_msg,
+            )
+            fail_report += f"{fail_report_run}\n"
+        delim = "*"
+        width = 80
+        msg = Template(_REPORT_FORMAT_TEMPLATE).substitute(
+            boarder=delim * width,
+            success_report=success_report or "<NONE>",
+            fail_report=fail_report or "<NONE>",
+        )
+        print(msg)
+        if len(failed_apps) > 0:
+            raise RuntimeError(
+                "Component test failed, see report above for detailed issue"
+            )
 
     def _get_app_def_run_status(
         self, app_def_run: AppDefRun, timeout: int
-    ) -> Tuple[str, str]:
+    ) -> Optional[str]:
         try:
-            print(f"Retrieving results from {app_def_run}: {app_def_run.provider_cls}")
-            app_name = app_def_run.future.result(timeout=timeout)
-            return "succeeded", f"`{app_name}`:`{app_def_run}`"
-        except Exception as e:
+            print(f"Retrieving {app_def_run.app_def.name}: {app_def_run.provider}")
+            app_def_run.future.result(timeout=timeout)
+            return None
+        except Exception:
             stack_trace_msg = traceback.format_exc().replace("\n", "\n  ")
-
-            msg = (
-                f"Failure while running: {app_def_run}, provider: `{app_def_run.provider_cls}`: {e}\n"
-                f"Stacktrace: {stack_trace_msg}\n"
-            )
-            return "failed", msg
+            return stack_trace_msg

torchx/components/integration_tests/test/__init__.py

@@ -0,0 +1,5 @@
+# Copyright (c) Facebook, Inc. and its affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.

torchx/components/integration_tests/test/dummy_app.py

@@ -0,0 +1,13 @@
+# Copyright (c) Facebook, Inc. and its affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+
+def main() -> None:
+    print("hi from main")
+
+
+if __name__ == "__main__":
+    main()