moving sidecar to simcore_service_sidecar

added RabbitMQ
updating from other project

Author: sanderegg

services/sidecar/setup.py

@@ -4,30 +4,55 @@
 
 from setuptools import find_packages, setup
 
-here = Path(sys.argv[0] if __name__ == "__main__" else __file__).resolve().parent
+if sys.version_info.major != 3 and sys.version_info.minor != 6:
+    raise RuntimeError(
+        "Expected ~=3.6, got %s. Did you forget to activate virtualenv?"
+        % str(sys.version_info)
+    )
+
+current_dir = Path(sys.argv[0] if __name__ == "__main__" else __file__).resolve().parent
 
 
 def read_reqs(reqs_path: Path):
     return re.findall(r"(^[^#-][\w]+[-~>=<.\w]+)", reqs_path.read_text(), re.MULTILINE)
 
+readme = (current_dir / "README.md").read_text()
+version = (current_dir / "VERSION").read_text().strip()
 
-install_requirements = read_reqs(here / "requirements" / "_base.txt") + [
+install_requirements = read_reqs(current_dir / "requirements" / "_base.txt") + [
     "s3wrapper==0.1.0",
     "simcore-sdk==0.1.0",
     "simcore-service-storage-sdk==0.1.0",
     "simcore-service-library",
 ]
 
-test_requirements = read_reqs(here / "requirements" / "_test.txt")
+test_requirements = read_reqs(current_dir / "requirements" / "_test.txt")
 
 
 setup(
     name="simcore-service-sidecar",
-    version="0.0.1",
+    version=version,
+    author="oSparclers",
+    description="Platform's sidecar",
+    classifiers={
+        "Development Status :: 1 - Planning",
+        "License :: OSI Approved :: MIT License",
+        "Natural Language :: English",
+        "Programming Language :: Python :: 3.6",
+    },
+    long_description=readme,
+    license="MIT license",
+    python_requires="~=3.6",
     packages=find_packages(where="src"),
-    package_dir={"": "src"},
+    package_dir={"": "src",},
+    include_package_data=True,
     install_requires=install_requirements,
-    python_requires=">=3.6",
     test_suite="tests",
     tests_require=test_requirements,
+    extras_require={"test": test_requirements},
+    entry_points={
+        "console_scripts": [
+            "simcore-service-sidecar = simcore_service_sidecar.__main__:main",
+        ],
+    },
 )

services/sidecar/src/sidecar/__init__.py

@@ -0,0 +1,3 @@
+""" Python package for the simcore_service_sidecar.
+"""
+from .__version__ import __version__

services/sidecar/src/simcore_service_sidecar/__init__.py

@@ -0,0 +1,43 @@
+""" Main application entry point
+ `python -m simcore_service_sidecar ...`
+Why does this file exist, and why __main__? For more info, read:
+- https://www.python.org/dev/peps/pep-0338/
+- https://docs.python.org/3/using/cmdline.html#cmdoption-m
+"""
+
+import logging
+
+import click
+
+from .utils import wrap_async_call
+
+log = logging.getLogger(__name__)
+
+@click.command
+@click.option("--job_id", default="0", help="The job ID")
+@click.option("--user_id", default="0", help="The user ID")
+@click.option("--project_id", default="0", help="The project ID")
+@click.option("--node_id", default=None, help="The node ID or nothing")
+def main(job_id: str, user_id: str, project_id: str, node_id: str):
+    from simcore_service_sidecar.core import SIDECAR
+
+    log.info(
+        "STARTING task processing for user %s, project %s, node %s",
+        user_id,
+        project_id,
+        node_id,
+    )
+    try:
+        next_task_nodes =  wrap_async_call(SIDECAR.inspect(job_id, user_id, project_id, node_id=node_id))
+    except Exception: # pylint: disable=broad-except
+        log.exception("Uncaught exception")
+    log.info(
+        "COMPLETED task processing for user %s, project %s, node %s",
+        user_id,
+        project_id,
+        node_id,
+    )
+
+
+if __name__ == "__main__":
+    main()

services/sidecar/src/simcore_service_sidecar/__main__.py

@@ -0,0 +1,10 @@
+""" Current version of the simcore_service_api_gateway application
+"""
+import pkg_resources
+
+__version__ = pkg_resources.get_distribution("simcore_service_sidecar").version
+
+major, minor, patch = __version__.split(".")
+
+api_version = __version__
+api_version_prefix: str = f"v{major}"

services/sidecar/src/simcore_service_sidecar/__version__.py

@@ -6,36 +6,27 @@
 from contextlib import contextmanager
 from datetime import datetime
 from pathlib import Path
-from typing import Dict, List, Union
+from typing import Dict, List, Union, Any
 
+import aio_pika
 import aiofiles
+import attr
 import docker
-import pika
 from celery.utils.log import get_task_logger
 from sqlalchemy import and_, exc
 
 from servicelib.utils import logged_gather
 from simcore_sdk import node_data, node_ports
-from simcore_sdk.models.pipeline_models import (
-    RUNNING,
-    SUCCESS,
-    ComputationalPipeline,
-    ComputationalTask,
-)
+from simcore_sdk.models.pipeline_models import (RUNNING, SUCCESS,
+                                                ComputationalPipeline,
+                                                ComputationalTask)
 from simcore_sdk.node_ports import log as node_port_log
 from simcore_sdk.node_ports.dbmanager import DBManager
 
 from . import config
-from .utils import (
-    DbSettings,
-    DockerSettings,
-    ExecutorSettings,
-    RabbitSettings,
-    S3Settings,
-    find_entry_point,
-    is_node_ready,
-    safe_channel,
-)
+from .rabbitmq import RabbitMQ
+from .utils import (DbSettings, DockerSettings, ExecutorSettings, S3Settings,
+                    find_entry_point, is_node_ready, safe_channel)
 
 log = get_task_logger(__name__)
 log.setLevel(config.SIDECAR_LOGLEVEL)
@@ -57,35 +48,17 @@ def session_scope(session_factory):
     finally:
         session.close()
 
-
+@attr.s(auto_attribs=True)
 class Sidecar:  # pylint: disable=too-many-instance-attributes
-    def __init__(self):
-        # publish subscribe config
-        self._pika = RabbitSettings()
-
-        # docker client config
-        self._docker = DockerSettings()
-
-        # object storage config
-        self._s3 = S3Settings()
-
-        # db config
-        self._db = DbSettings()  # keeps single db engine: sidecar.utils_{id}
-        self._db_manager = (
-            None  # lazy init because still not configured. SEE _get_node_ports
-        )
-
-        # current task
-        self._task = None
-
-        # current user id
-        self._user_id: str = None
-
-        # stack name
-        self._stack_name: str = None
-
-        # executor options
-        self._executor = ExecutorSettings()
+    _rabbit_mq: RabbitMQ
+    _docker: DockerSettings = DockerSettings()
+    _s3: S3Settings = S3Settings()
+    _db: DbSettings = DbSettings() # keeps single db engine: sidecar.utils_{id}
+    _db_manager: Any = None # lazy init because still not configured. SEE _get_node_ports
+    _task: ComputationalTask = None # current task
+    _user_id: str = None # current user id
+    _stack_name: str = None # stack name
+    _executor: ExecutorSettings = ExecutorSettings() # executor options
 
     async def _get_node_ports(self):
         if self._db_manager is None:
@@ -211,27 +184,51 @@ async def _post_progress(self, channel, progress):
     async def log_file_processor(self, log_file: Path) -> None:
         """checks both container logs and the log_file if any
         """
-        try:
-            TIME_BETWEEN_LOGS_S: int = 2
-            time_logs_sent = time.monotonic()
-            accumulated_logs = []
-            async with aiofiles.open(log_file, mode="r") as fp:
-                async for line in fp:
-                    now = time.monotonic()
-                    accumulated_logs.append(line)
-                    if (now - time_logs_sent) < TIME_BETWEEN_LOGS_S:
-                        continue
-                    # send logs to rabbitMQ
-                    # TODO: NEEDS to shield??
-                    with safe_channel(self._pika) as (channel, _):
-                        await self._post_log(channel, msg=accumulated_logs)
-                        time_logs_sent = now
-                        accumulated_logs = []
-        except asyncio.CancelledError:
-            # the task is complete let's send the last logs
-            if accumulated_logs:
-                with safe_channel(self._pika) as (channel, _):
-                    await self._post_log(channel, msg=accumulated_logs)
+        # async def parse_line(line: str) -> None:
+        #     # TODO: This should be 'settings', a regex for every service
+        #     if line.lower().startswith("[progress]"):
+        #         progress = line.lower().lstrip(
+        #             "[progress]").rstrip("%").strip()
+        #         await self._post_progress(channel, progress)
+        #         log.debug('PROGRESS %s', progress)
+        #     elif "percent done" in line.lower():
+        #         progress = line.lower().rstrip("percent done")
+        #         try:
+        #             float_progress = float(progress) / 100.0
+        #             progress = str(float_progress)
+        #             await self._post_progress(channel, progress)
+        #             log.debug('PROGRESS %s', progress)
+        #         except ValueError:
+        #             log.exception("Could not extract progress from solver")
+        #     else:
+        #         # just send as log
+        #         await self._post_log(channel, msg=line)
+
+        
+
+
+        # try:
+        #     import pdb; pdb.set_trace()
+        #     TIME_BETWEEN_LOGS_S: int = 2
+        #     time_logs_sent = time.monotonic()
+        #     accumulated_logs = []
+        #     async with aiofiles.open(log_file, mode="r") as fp:
+        #         async for line in fp:
+        #             now = time.monotonic()
+        #             accumulated_logs.append(line)
+        #             if (now - time_logs_sent) < TIME_BETWEEN_LOGS_S:
+        #                 continue
+        #             # send logs to rabbitMQ
+        #             # TODO: NEEDS to shield??
+        #             with safe_channel(self._pika) as (channel, _):
+        #                 await self._post_log(channel, msg=accumulated_logs)
+        #                 time_logs_sent = now
+        #                 accumulated_logs = []
+        # except asyncio.CancelledError:
+        #     # the task is complete let's send the last logs
+        #     if accumulated_logs:
+        #         with safe_channel(self._pika) as (channel, _):
+        #             await self._post_log(channel, msg=accumulated_logs)
 
     # async def _bg_job(self, log_file):
     #     log.debug('Bck job started %s:node %s:internal id %s from container',
@@ -568,7 +565,7 @@ async def postprocess(self):
         finally:
             _session.close()
 
-    async def inspect(self, celery_task, user_id: str, project_id: str, node_id: str):
+    async def inspect(self, job_request_id: int, user_id: str, project_id: str, node_id: str):
         log.debug(
             "ENTERING inspect with user %s pipeline:node %s: %s",
             user_id,
@@ -657,4 +654,3 @@ async def inspect(self, celery_task, user_id: str, project_id: str, node_id: str
 SIDECAR = Sidecar()
 
 __all__ = ["SIDECAR"]
-

services/sidecar/src/sidecar/_deprecated.py renamed to services/sidecar/src/simcore_service_sidecar/_deprecated.py

@@ -0,0 +1,46 @@
+import logging
+
+import aio_pika
+import attr
+import tenacity
+
+from servicelib.rabbitmq_utils import RabbitMQRetryPolicyUponInitialization
+from simcore_sdk.config.rabbit import Config, eval_broker
+
+log = logging.getLogger(__file__)
+
+@attr.s(auto_attribs=True)
+class RabbitMQ:
+    _config: Config = Config()
+    _connection: aio_pika.RobustConnection = None
+    _channel: aio_pika.Channel = None
+    logs_exchange: aio_pika.Exchange = None    
+    progress_exchange: aio_pika.Exchange = None
+
+    async def connect(self):
+        url = eval_broker(self._config)
+        await wait_till_rabbit_responsive(url)
+
+
+        self._connection = await aio_pika.connect_robust(
+            url,
+            client_properties={"connection_name": "sidecar connection"},
+        )
+
+        self._channel = await self._connection.channel()
+        self.logs_exchange = await self._channel.declare_exchange(
+            self._config.log_channel, aio_pika.ExchangeType.FANOUT, auto_delete=True
+        )
+        self.progress_exchange = await self._channel.declare_exchange(
+            self._config.progress_channel, aio_pika.ExchangeType.FANOUT, auto_delete=True
+        )
+
+    async def post_message(self):
+        
+
+
+@tenacity.retry(**RabbitMQRetryPolicyUponInitialization().kwargs)
+async def wait_till_rabbit_responsive(url: str):
+    connection = await aio_pika.connect(url)
+    await connection.close()
+    return True

services/sidecar/src/sidecar/celery.py renamed to services/sidecar/src/simcore_service_sidecar/celery.py

@@ -13,7 +13,7 @@ async def async_pipeline(self, user_id: str, project_id: str, node_id: str) -> L
     from .core import SIDECAR
 
     log.info("STARTING task processing for user %s, project %s, node %s", user_id, project_id, node_id)
-    next_task_nodes = await SIDECAR.inspect(self, user_id, project_id, node_id)
+    next_task_nodes = await SIDECAR.inspect(self.request.id, user_id, project_id, node_id)
     log.info("COMPLETED task processing for user %s, project %s, node %s", user_id, project_id, node_id)
     return next_task_nodes