Modified elasticsearch file name to 'elastic' to avoid name errors. Applied changes suggested in the pr.

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']
 

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",
     )

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,
     )
 
 

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,