diff --git a/piker/data/_ahab.py b/piker/data/_ahab.py index d7113b61..1964eb1b 100644 --- a/piker/data/_ahab.py +++ b/piker/data/_ahab.py @@ -124,6 +124,8 @@ class Container: async def process_logs_until( self, + # this is a predicate func for matching log msgs emitted by the + # underlying containerized app patt_matcher: Callable[[str], bool], bp_on_msg: bool = False, ) -> bool: @@ -152,7 +154,7 @@ class Container: elif 'message' in record: msg = record['message'] else: - raise KeyError('Unexpected log format') + raise KeyError(f'Unexpected log format\n{record}') level = record['level'] diff --git a/piker/data/elasticsearch.py b/piker/data/elastic.py similarity index 83% rename from piker/data/elasticsearch.py rename to piker/data/elastic.py index c3344aa3..f9ed7b16 100644 --- a/piker/data/elasticsearch.py +++ b/piker/data/elastic.py @@ -25,43 +25,37 @@ ''' from __future__ import annotations from contextlib import asynccontextmanager as acm -from datetime import datetime from pprint import pformat from typing import ( Any, - Optional, - Union, + # Optional, + #Union, TYPE_CHECKING, ) -import time -from math import isnan -from bidict import bidict -from msgspec.msgpack import encode, decode import pyqtgraph as pg import numpy as np import tractor -from trio_websocket import open_websocket_url -import pendulum -import purerpc + if TYPE_CHECKING: import docker from ._ahab import DockerContainer -from .feed import maybe_open_feed -from ..log import get_logger, get_console_log -from .._profile import Profiler +from piker.log import ( + get_logger, + get_console_log +) from elasticsearch import Elasticsearch -from docker.types import LogConfig + log = get_logger(__name__) # container level config _config = { - 'port': 9200, + 'port': 19200, 'log_level': 'debug', } @@ -103,6 +97,6 @@ def start_elasticsearch( dcntr, {}, # expected startup and stop msgs - lambda msg: msg == "started", - lambda msg: msg == "closed", + lambda start_msg: start_msg == "started", + lambda stop_msg: stop_msg == "closed", ) diff --git a/piker/data/marketstore.py b/piker/data/marketstore.py index 2595a2d6..4a6131b2 100644 --- a/piker/data/marketstore.py +++ b/piker/data/marketstore.py @@ -194,8 +194,8 @@ def start_marketstore( _config, # expected startup and stop msgs - lambda msg: "launching tcp listener for all services..." in msg, - lambda msg: "exiting..." in msg, + lambda start_msg: "launching tcp listener for all services..." in start_msg, + lambda stop_msg: "exiting..." in stop_msg, ) diff --git a/tests/conftest.py b/tests/conftest.py index 2cfaad7a..9ccc11ab 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -119,6 +119,8 @@ def cse_symbols(): @acm async def _open_test_pikerd( + tsdb: bool = False, + es: bool = False, reg_addr: tuple[str, int] | None = None, **kwargs, @@ -143,6 +145,8 @@ async def _open_test_pikerd( # try: async with ( maybe_open_pikerd( + tsdb=tsdb, + es=es, registry_addr=reg_addr, **kwargs, ) as service_manager,