piker/__init__.py

@@ -1,5 +1,5 @@
 # piker: trading gear for hackers.
-# Copyright 2020-eternity Tyler Goodlet (in stewardship for pikers)
+# Copyright 2020-eternity Tyler Goodlet (in stewardship for piker0)
 
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as published by
@@ -14,11 +14,11 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
-'''
+"""
 piker: trading gear for hackers.
 
-'''
+"""
-from .service import open_piker_runtime
+from ._daemon import open_piker_runtime
 from .data.feed import open_feed
 
 __all__ = [

piker/_daemon.py

@@ -19,8 +19,6 @@ Structured, daemon tree service management.
 
 """
 from __future__ import annotations
-from pprint import pformat
-from functools import partial
 import os
 from typing import (
     Optional,
@@ -37,11 +35,14 @@ import tractor
 import trio
 from trio_typing import TaskStatus
 
-from ..log import (
+from .log import (
     get_logger,
     get_console_log,
 )
-from ..brokers import get_brokermod
+from .brokers import get_brokermod
+
+from pprint import pformat
+from functools import partial
 
 
 log = get_logger(__name__)
@@ -336,6 +337,7 @@ async def open_pikerd(
     alive underling services (see below).
 
     '''
+
     async with (
         open_piker_runtime(
 
@@ -353,26 +355,17 @@ async def open_pikerd(
         tractor.open_nursery() as actor_nursery,
         trio.open_nursery() as service_nursery,
     ):
-        if root_actor.accept_addr != reg_addr:
+        assert root_actor.accept_addr == reg_addr
-            raise RuntimeError(f'Daemon failed to bind on {reg_addr}!?')
-
-        # assign globally for future daemon/task creation
-        Services.actor_n = actor_nursery
-        Services.service_n = service_nursery
-        Services.debug_mode = debug_mode
 
         if tsdb:
-            from ._ahab import start_ahab
+            from piker.data._ahab import start_ahab
-            from .marketstore import start_marketstore
+            from piker.data.marketstore import start_marketstore
 
             log.info('Spawning `marketstore` supervisor')
             ctn_ready, config, (cid, pid) = await service_nursery.start(
-                partial(
+                start_ahab,
-                    start_ahab,
+                'marketstored',
-                    'marketstored',
+                start_marketstore,
-                    start_marketstore,
-                    loglevel=loglevel,
-                )
 
             )
             log.info(
@@ -392,7 +385,7 @@ async def open_pikerd(
                     start_ahab,
                     'elasticsearch',
                     start_elasticsearch,
-                    loglevel=loglevel,
+                    start_timeout=240.0  # high cause ci
                 )
             )
 
@@ -403,6 +396,12 @@ async def open_pikerd(
                 f'config: {pformat(config)}'
             )
 
+        # assign globally for future daemon/task creation
+        Services.actor_n = actor_nursery
+        Services.service_n = service_nursery
+        Services.debug_mode = debug_mode
+
+
         try:
             yield Services
 
@@ -668,7 +667,7 @@ async def spawn_brokerd(
     )
 
     # non-blocking setup of brokerd service nursery
-    from ..data import _setup_persistent_brokerd
+    from .data import _setup_persistent_brokerd
 
     await Services.start_service_task(
         dname,
@@ -696,10 +695,7 @@ async def maybe_spawn_brokerd(
 
         f'brokerd.{brokername}',
         service_task_target=spawn_brokerd,
-        spawn_args={
+        spawn_args={'brokername': brokername, 'loglevel': loglevel},
-            'brokername': brokername,
-            'loglevel': loglevel,
-        },
         loglevel=loglevel,
         **kwargs,
 
@@ -731,7 +727,7 @@ async def spawn_emsd(
     )
 
     # non-blocking setup of clearing service
-    from ..clearing._ems import _setup_persistent_emsd
+    from .clearing._ems import _setup_persistent_emsd
 
     await Services.start_service_task(
         'emsd',

piker/brokers/cli.py

@@ -29,15 +29,8 @@ import tractor
 from ..cli import cli
 from .. import watchlists as wl
 from ..log import get_console_log, colorize_json, get_logger
-from ..service import (
+from .._daemon import maybe_spawn_brokerd, maybe_open_pikerd
-    maybe_spawn_brokerd,
+from ..brokers import core, get_brokermod, data
-    maybe_open_pikerd,
-)
-from ..brokers import (
-    core,
-    get_brokermod,
-    data,
-)
 
 log = get_logger('cli')
 DEFAULT_BROKER = 'questrade'
@@ -67,7 +60,6 @@ def get_method(client, meth_name: str):
     print_ok('found!.')
     return method
 
-
 async def run_method(client, meth_name: str, **kwargs):
     method = get_method(client, meth_name)
     print('running...', end='', flush=True)
@@ -75,20 +67,19 @@ async def run_method(client, meth_name: str, **kwargs):
     print_ok(f'done! result: {type(result)}')
     return result
 
-
 async def run_test(broker_name: str):
     brokermod = get_brokermod(broker_name)
     total = 0
     passed = 0
     failed = 0
 
-    print('getting client...', end='', flush=True)
+    print(f'getting client...', end='', flush=True)
     if not hasattr(brokermod, 'get_client'):
         print_error('fail! no \'get_client\' context manager found.')
         return
 
     async with brokermod.get_client(is_brokercheck=True) as client:
-        print_ok('done! inside client context.')
+        print_ok(f'done! inside client context.')
 
         # check for methods present on brokermod
         method_list = [
@@ -139,6 +130,7 @@ async def run_test(broker_name: str):
 
             total += 1
 
+
         # check for methods present con brokermod.Client and their
         # results
 
@@ -188,6 +180,7 @@ def brokercheck(config, broker):
     trio.run(run_test, broker)
 
 
+
 @cli.command()
 @click.option('--keys', '-k', multiple=True,
               help='Return results only for these keys')
@@ -342,6 +335,8 @@ def contracts(ctx, loglevel, broker, symbol, ids):
     brokermod = get_brokermod(broker)
     get_console_log(loglevel)
 
+
+
     contracts = trio.run(partial(core.contracts, brokermod, symbol))
     if not ids:
         # just print out expiry dates which can be used with

piker/brokers/core.py

@@ -28,7 +28,7 @@ import trio
 
 from ..log import get_logger
 from . import get_brokermod
-from ..service import maybe_spawn_brokerd
+from .._daemon import maybe_spawn_brokerd
 from .._cacheables import open_cached_client
 
 

piker/clearing/_client.py

@@ -29,11 +29,8 @@ from tractor.trionics import broadcast_receiver
 
 from ..log import get_logger
 from ..data.types import Struct
-from ..service import maybe_open_emsd
+from .._daemon import maybe_open_emsd
-from ._messages import (
+from ._messages import Order, Cancel
-    Order,
-    Cancel,
-)
 from ..brokers import get_brokermod
 
 if TYPE_CHECKING:

piker/cli/__init__.py

@@ -19,18 +19,16 @@ CLI commons.
 
 '''
 import os
+from pprint import pformat
+from functools import partial
 
 import click
 import trio
 import tractor
 
-from ..log import (
+from ..log import get_console_log, get_logger, colorize_json
-    get_console_log,
-    get_logger,
-    colorize_json,
-)
 from ..brokers import get_brokermod
-from ..service import (
+from .._daemon import (
     _default_registry_host,
     _default_registry_port,
 )
@@ -70,7 +68,7 @@ def pikerd(
 
     '''
 
-    from ..service import open_pikerd
+    from .._daemon import open_pikerd
     log = get_console_log(loglevel)
 
     if pdb:
@@ -173,7 +171,7 @@ def cli(
 @click.pass_obj
 def services(config, tl, ports):
 
-    from ..service import (
+    from .._daemon import (
         open_piker_runtime,
         _default_registry_port,
         _default_registry_host,
@@ -206,8 +204,8 @@ def services(config, tl, ports):
 
 
 def _load_clis() -> None:
-    from ..service import marketstore  # noqa
+    from ..data import marketstore  # noqa
-    from ..service import elastic
+    from ..data import elastic
     from ..data import cli  # noqa
     from ..brokers import cli  # noqa
     from ..ui import cli  # noqa

piker/data/_ahab.py

@@ -19,7 +19,6 @@ Supervisor for docker with included specific-image service helpers.
 
 '''
 from collections import ChainMap
-from functools import partial
 import os
 import time
 from typing import (
@@ -47,10 +46,7 @@ from requests.exceptions import (
     ReadTimeout,
 )
 
-from ..log import (
+from ..log import get_logger, get_console_log
-    get_logger,
-    get_console_log,
-)
 from .. import config
 
 log = get_logger(__name__)
@@ -201,11 +197,6 @@ class Container:
 
         return False
 
-    @property
-    def cuid(self) -> str:
-        fqcn: str = self.cntr.attrs['Config']['Image']
-        return f'{fqcn}[{self.cntr.short_id}]'
-
     def try_signal(
         self,
         signal: str = 'SIGINT',
@@ -241,23 +232,17 @@ class Container:
 
     async def cancel(
         self,
-        log_msg_key: str,
+        stop_msg: str,
-        stop_predicate: Callable[[str], bool],
-
         hard_kill: bool = False,
 
     ) -> None:
-        '''
-        Attempt to cancel this container gracefully, fail over to
-        a hard kill on timeout.
 
-        '''
         cid = self.cntr.id
 
         # first try a graceful cancel
         log.cancel(
-            f'SIGINT cancelling container: {self.cuid}\n'
+            f'SIGINT cancelling container: {cid}\n'
-            'waiting on stop predicate...'
+            f'waiting on stop msg: "{stop_msg}"'
         )
         self.try_signal('SIGINT')
 
@@ -268,10 +253,7 @@ class Container:
                 log.cancel('polling for CNTR logs...')
 
                 try:
-                    await self.process_logs_until(
+                    await self.process_logs_until(stop_msg)
-                        log_msg_key,
-                        stop_predicate,
-                    )
                 except ApplicationLogError:
                     hard_kill = True
                 else:
@@ -329,16 +311,11 @@ class Container:
 async def open_ahabd(
     ctx: tractor.Context,
     endpoint: str,  # ns-pointer str-msg-type
-    loglevel: str | None = 'cancel',
 
     **kwargs,
 
 ) -> None:
-
+    get_console_log('info', name=__name__)
-    log = get_console_log(
-        loglevel,
-        name=__name__,
-    )
 
     async with open_docker() as client:
 
@@ -361,63 +338,40 @@ async def open_ahabd(
 
             # defaults
             {
-                # startup time limit which is the max the supervisor
-                # will wait for the container to be registered in
-                # ``client.containers.list()``
                 'startup_timeout': 1.0,
-
-                # how fast to poll for the starup predicate by sleeping
-                # this amount incrementally thus yielding to the
-                # ``trio`` scheduler on during sync polling execution.
                 'startup_query_period': 0.001,
-
-                # str-key value expected to contain log message body-contents
-                # when read using:
-                # ``json.loads(entry for entry in DockerContainer.logs())``
                 'log_msg_key': 'msg',
             },
         )
 
-        with trio.move_on_after(conf['startup_timeout']) as cs:
+        found = False
-            async with trio.open_nursery() as tn:
+        with trio.move_on_after(conf['startup_timeout']):
-                tn.start_soon(
+            found = await cntr.process_logs_until(
-                    partial(
+                conf['log_msg_key'],
-                        cntr.process_logs_until,
+                start_lambda,
-                        log_msg_key=conf['log_msg_key'],
+                checkpoint_period=conf['startup_query_period'],
-                        patt_matcher=start_lambda,
+            )
-                        checkpoint_period=conf['startup_query_period'],
-                    )
-                )
 
-                # poll for container startup or timeout
+        # XXX: if we timeout on finding the "startup msg" we expect then
-                while not cs.cancel_called:
+        # we want to FOR SURE raise an error upwards!
-                    if dcntr in client.containers.list():
+        if (
-                        break
+            not found
+            and dcntr not in client.containers.list()
+        ):
+            for entry in cntr.seen_so_far:
+                log.info(entry)
 
-                    await trio.sleep(conf['startup_query_period'])
+            raise RuntimeError(
+                f'Failed to start {dcntr.id} check logs deats'
+            )
 
-                # sync with remote caller actor-task but allow log
+        await ctx.started((
-                # processing to continue running in bg.
+            cntr.cntr.id,
-                await ctx.started((
+            os.getpid(),
-                    cntr.cntr.id,
+            cntr_config,
-                    os.getpid(),
+        ))
-                    cntr_config,
-                ))
 
         try:
-            # XXX: if we timeout on finding the "startup msg" we expect then
-            # we want to FOR SURE raise an error upwards!
-            if cs.cancelled_caught:
-                # if dcntr not in client.containers.list():
-                for entry in cntr.seen_so_far:
-                    log.info(entry)
-
-                raise DockerNotStarted(
-                    f'Failed to start container: {cntr.cuid}\n'
-                    f'due to startup_timeout={conf["startup_timeout"]}s\n\n'
-                    "prolly you should check your container's logs for deats.."
-                )
-
             # TODO: we might eventually want a proxy-style msg-prot here
             # to allow remote control of containers without needing
             # callers to have root perms?
@@ -426,21 +380,14 @@ async def open_ahabd(
         finally:
             # TODO: ensure loglevel can be set and teardown logs are
             # reported if possible on error or cancel..
-            # XXX WARNING: currently shielding here can result in hangs
-            # on ctl-c from user.. ideally we can avoid a cancel getting
-            # consumed and not propagating whilst still doing teardown
-            # logging..
             # with trio.CancelScope(shield=True):
-            await cntr.cancel(
+            await cntr.cancel(stop_lambda)
-                log_msg_key=conf['log_msg_key'],
-                stop_predicate=stop_lambda,
-            )
 
 
 async def start_ahab(
     service_name: str,
     endpoint: Callable[docker.DockerClient, DockerContainer],
-    loglevel: str | None = 'cancel',
+    loglevel: str | None = None,
 
     task_status: TaskStatus[
         tuple[
@@ -462,12 +409,13 @@ async def start_ahab(
     '''
     cn_ready = trio.Event()
     try:
-        async with tractor.open_nursery() as an:
+        async with tractor.open_nursery(
+            loglevel='runtime',
+        ) as tn:
 
-            portal = await an.start_actor(
+            portal = await tn.start_actor(
                 service_name,
-                enable_modules=[__name__],
+                enable_modules=[__name__]
-                loglevel=loglevel,
             )
 
             # TODO: we have issues with this on teardown
@@ -488,7 +436,6 @@ async def start_ahab(
             async with portal.open_context(
                 open_ahabd,
                 endpoint=str(NamespacePath.from_ref(endpoint)),
-                loglevel='cancel',
             ) as (ctx, first):
 
                 cid, pid, cntr_config = first

piker/data/_sampling.py

@@ -42,7 +42,7 @@ from ..log import (
     get_logger,
     get_console_log,
 )
-from ..service import maybe_spawn_daemon
+from .._daemon import maybe_spawn_daemon
 
 if TYPE_CHECKING:
     from ._sharedmem import (
@@ -69,7 +69,7 @@ class Sampler:
     This non-instantiated type is meant to be a singleton within
     a `samplerd` actor-service spawned once by the user wishing to
     time-step-sample (real-time) quote feeds, see
-    ``.service.maybe_open_samplerd()`` and the below
+    ``._daemon.maybe_open_samplerd()`` and the below
     ``register_with_sampler()``.
 
     '''
@@ -391,7 +391,7 @@ async def spawn_samplerd(
     update and increment count write and stream broadcasting.
 
     '''
-    from piker.service import Services
+    from piker._daemon import Services
 
     dname = 'samplerd'
     log.info(f'Spawning `{dname}`')

piker/data/cli.py

@@ -137,14 +137,14 @@ def storesh(
 
     '''
     from piker.data.marketstore import open_tsdb_client
-    from piker.service import open_piker_runtime
+    from piker._daemon import open_piker_runtime
 
     async def main():
         nonlocal symbols
 
         async with open_piker_runtime(
             'storesh',
-            enable_modules=['piker.service._ahab'],
+            enable_modules=['piker.data._ahab'],
         ):
             symbol = symbols[0]
 
@@ -187,14 +187,14 @@ def storage(
 
     '''
     from piker.data.marketstore import open_tsdb_client
-    from piker.service import open_piker_runtime
+    from piker._daemon import open_piker_runtime
 
     async def main():
         nonlocal symbols
 
         async with open_piker_runtime(
             'tsdb_storage',
-            enable_modules=['piker.service._ahab'],
+            enable_modules=['piker.data._ahab'],
         ):
             symbol = symbols[0]
             async with open_tsdb_client(symbol) as storage:

piker/data/elastic.py

@@ -15,11 +15,17 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 from __future__ import annotations
+from contextlib import asynccontextmanager as acm
+from pprint import pformat
 from typing import (
     Any,
     TYPE_CHECKING,
 )
 
+import pyqtgraph as pg
+import numpy as np
+import tractor
+
 
 if TYPE_CHECKING:
     import docker
@@ -59,14 +65,14 @@ def start_elasticsearch(
             -itd \
             --rm \
             --network=host \
-            --mount type=bind,source="$(pwd)"/elastic,\
+            --mount type=bind,source="$(pwd)"/elastic,target=/usr/share/elasticsearch/data \
-              target=/usr/share/elasticsearch/data \
             --env "elastic_username=elastic" \
             --env "elastic_password=password" \
             --env "xpack.security.enabled=false" \
             elastic
 
     '''
+    import docker
     get_console_log('info', name=__name__)
 
     dcntr: DockerContainer = client.containers.run(
@@ -80,7 +86,7 @@ def start_elasticsearch(
     async def start_matcher(msg: str):
         try:
             health = (await asks.get(
-                'http://localhost:19200/_cat/health',
+                f'http://localhost:19200/_cat/health',
                 params={'format': 'json'}
             )).json()
 
@@ -96,17 +102,7 @@ def start_elasticsearch(
 
     return (
         dcntr,
-        {
+        {},
-            # apparently we're REALLY tolerant of startup latency
-            # for CI XD
-            'startup_timeout': 240.0,
-
-            # XXX: decrease http poll period bc docker
-            # is shite at handling fast poll rates..
-            'startup_query_period': 0.1,
-
-            'log_msg_key': 'message',
-        },
         # expected startup and stop msgs
         start_matcher,
         stop_matcher,

piker/data/feed.py

@@ -58,7 +58,7 @@ from ..log import (
     get_logger,
     get_console_log,
 )
-from ..service import (
+from .._daemon import (
     maybe_spawn_brokerd,
     check_for_service,
 )

piker/data/marketstore.py

@@ -26,6 +26,7 @@
 from __future__ import annotations
 from contextlib import asynccontextmanager as acm
 from datetime import datetime
+from pprint import pformat
 from typing import (
     Any,
     Optional,
@@ -54,7 +55,7 @@ if TYPE_CHECKING:
     import docker
     from ._ahab import DockerContainer
 
-from ..data.feed import maybe_open_feed
+from .feed import maybe_open_feed
 from ..log import get_logger, get_console_log
 from .._profile import Profiler
 
@@ -62,12 +63,11 @@ from .._profile import Profiler
 log = get_logger(__name__)
 
 
-# ahabd-supervisor and container level config
+# container level config
 _config = {
     'grpc_listen_port': 5995,
     'ws_listen_port': 5993,
     'log_level': 'debug',
-    'startup_timeout': 2,
 }
 
 _yaml_config = '''
@@ -135,7 +135,7 @@ def start_marketstore(
 
     # create dirs when dne
     if not os.path.isdir(config._config_dir):
-        Path(config._config_dir).mkdir(parents=True, exist_ok=True)
+       Path(config._config_dir).mkdir(parents=True, exist_ok=True)
 
     if not os.path.isdir(mktsdir):
         os.mkdir(mktsdir)

piker/log.py

@@ -21,11 +21,7 @@ import logging
 import json
 
 import tractor
-from pygments import (
+from pygments import highlight, lexers, formatters
-    highlight,
-    lexers,
-    formatters,
-)
 
 # Makes it so we only see the full module name when using ``__name__``
 # without the extra "piker." prefix.
@@ -36,48 +32,26 @@ def get_logger(
     name: str = None,
 
 ) -> logging.Logger:
-    '''
+    '''Return the package log or a sub-log for `name` if provided.
-    Return the package log or a sub-log for `name` if provided.
-
     '''
     return tractor.log.get_logger(name=name, _root_name=_proj_name)
 
 
-def get_console_log(
+def get_console_log(level: str = None, name: str = None) -> logging.Logger:
-    level: str | None = None,
+    '''Get the package logger and enable a handler which writes to stderr.
-    name: str | None = None,
-
-) -> logging.Logger:
-    '''
-    Get the package logger and enable a handler which writes to stderr.
 
     Yeah yeah, i know we can use ``DictConfig``. You do it...
-
     '''
     return tractor.log.get_console_log(
-        level,
+        level, name=name, _root_name=_proj_name)  # our root logger
-        name=name,
-        _root_name=_proj_name,
-    )  # our root logger
 
 
-def colorize_json(
+def colorize_json(data, style='algol_nu'):
-    data: dict,
+    """Colorize json output using ``pygments``.
-    style='algol_nu',
+    """
-):
+    formatted_json = json.dumps(data, sort_keys=True, indent=4)
-    '''
-    Colorize json output using ``pygments``.
-
-    '''
-    formatted_json = json.dumps(
-        data,
-        sort_keys=True,
-        indent=4,
-    )
     return highlight(
-        formatted_json,
+        formatted_json, lexers.JsonLexer(),
-        lexers.JsonLexer(),
-
         # likeable styles: algol_nu, tango, monokai
         formatters.TerminalTrueColorFormatter(style=style)
     )

piker/ui/_app.py

@@ -24,7 +24,7 @@ from types import ModuleType
 from PyQt5.QtCore import QEvent
 import trio
 
-from ..service import maybe_spawn_brokerd
+from .._daemon import maybe_spawn_brokerd
 from . import _event
 from ._exec import run_qtractor
 from ..data.feed import install_brokerd_search

piker/ui/_exec.py

@@ -49,7 +49,7 @@ from qdarkstyle import DarkPalette
 import trio
 from outcome import Error
 
-from ..service import (
+from .._daemon import (
     maybe_open_pikerd,
     get_tractor_runtime_kwargs,
 )

piker/ui/cli.py

@@ -24,7 +24,7 @@ import tractor
 
 from ..cli import cli
 from .. import watchlists as wl
-from ..service import maybe_spawn_brokerd
+from .._daemon import maybe_spawn_brokerd
 
 
 _config_dir = click.get_app_dir('piker')

tests/conftest.py

@@ -1,6 +1,7 @@
 from contextlib import asynccontextmanager as acm
 from functools import partial
 import os
+from typing import AsyncContextManager
 from pathlib import Path
 from shutil import rmtree
 
@@ -10,7 +11,7 @@ from piker import (
     # log,
     config,
 )
-from piker.service import (
+from piker._daemon import (
     Services,
 )
 from piker.clearing._client import open_ems
@@ -87,7 +88,7 @@ async def _open_test_pikerd(
 
     '''
     import random
-    from piker.service import maybe_open_pikerd
+    from piker._daemon import maybe_open_pikerd
 
     if reg_addr is None:
         port = random.randint(6e3, 7e3)
@@ -150,9 +151,8 @@ async def _open_test_pikerd_and_ems(
             fqsn,
             mode=mode,
             loglevel=loglevel,
-        ) as ems_services,
+        ) as ems_services):
-    ):
+            yield (services, ems_services)
-        yield (services, ems_services)
 
 
 @pytest.fixture
@@ -168,7 +168,7 @@ def open_test_pikerd_and_ems(
         mode,
         loglevel,
         open_test_pikerd
-    )
+    ) 
 
 
 @pytest.fixture(scope='module')

tests/test_databases.py

@@ -3,7 +3,7 @@ import trio
 
 from typing import AsyncContextManager
 
-from piker.service import Services
+from piker._daemon import Services
 from piker.log import get_logger
 
 from elasticsearch import Elasticsearch

tests/test_services.py

@@ -9,7 +9,8 @@ import pytest
 import trio
 import tractor
 
-from piker.service import (
+from piker.log import get_logger
+from piker._daemon import (
     find_service,
     Services,
 )