add support for reading .j2 config files with jinja

.github/workflows/main.yaml

@@ -16,12 +16,12 @@ jobs:
       - name: Set up Python 3.14
         uses: actions/setup-python@v6
         with:
-          python-version: 3.14
+          python-version: "3.14"
       - name: Install black
         run: pip install black
 
       - name: Check code with black
-        run: black --check --exclude kotekan .
+        run: black --check --exclude kotekan coco
 
       - name: Run black on scripts
         run: black --check scripts/cocod scripts/coco
@@ -34,7 +34,7 @@ jobs:
       - name: Set up Python 3.14
         uses: actions/setup-python@v6
         with:
-          python-version: 3.14
+          python-version: "3.14"
       - name: Install pydocstyle
         run: pip install pydocstyle
       - name: Run pydocstyle on all .py files
@@ -48,7 +48,7 @@ jobs:
       - name: Set up Python 3.14
         uses: actions/setup-python@v6
         with:
-          python-version: 3.14
+          python-version: "3.14"
       - name: Install flake8
         run: pip install flake8
       - name: Run flake8
@@ -62,7 +62,7 @@ jobs:
       - name: Set up Python
         uses: actions/setup-python@v6
         with:
-          python-version: "3.10"
+          python-version: "3.14"
       - name: Install pylint
         run: pip install pylint==2.15 pylint-ignore
       - name: Install dependencies
@@ -72,16 +72,20 @@ jobs:
 
   tests:
     runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python_version: ['3.8', '3.9', '3.10', '3.11']
     steps:
       - uses: actions/checkout@v6
 
       - name: Set up Python
         uses: actions/setup-python@v6
         with:
-          python-version: "3.10"
+          python-version: ${{ matrix.python_version }}
 
       - name: Install apt dependencies
         run: |
+          sudo apt-get update
           sudo apt-get install -y python3-sphinx libevent-dev libhdf5-dev g++
       - name: Install pip dependencies
         run: |
@@ -90,7 +94,7 @@ jobs:
           pip install .
 
       - name: Start Redis
-        uses: supercharge/redis-github-action@1.2.0
+        uses: supercharge/redis-github-action@1.4.0
         with:
           redis-version: 5
 

Dockerfile

@@ -1,27 +1,22 @@
 # Use an official Python runtime as a base image
-FROM python:3.7-slim
+FROM python:3.10-slim
 
 ## The maintainer name and email
 LABEL maintainer="CHIME/FRB Collaboration"
 
 ADD . /coco
 
 RUN apt-get update && \
-    apt-get install -y apt-utils && \
-    apt-get install -y software-properties-common && \
-    apt-get install -y git && \
-    apt-get install -y build-essential && \
-    apt-get install -y libmariadb-dev && \
-    apt-get install -y libevent-dev && \
-    apt-get install -y libhdf5-dev && \
-    pip install flask && \
-    pip install -r /coco/requirements.txt && \
-    pip install /coco && \
+    apt-get install -y apt-utils git build-essential curl \
+    libmariadb-dev libevent-dev && \
+    pip install --use-deprecated=legacy-resolver flask && \
+    pip install --use-deprecated=legacy-resolver -r /coco/requirements.txt && \
+    pip install --use-deprecated=legacy-resolver /coco
 
-    #-----------------------
-    # Minimize container size
-    #-----------------------
-    apt-get remove -y curl git && \
+#-----------------------
+# Minimize container size
+#-----------------------
+RUN apt-get remove -y git && \
     apt-get autoremove -y && \
     apt-get clean -y && \
-    rm -rf /tmp/build
+    rm -rf /tmp/build /coco

coco/_version.py

@@ -56,7 +56,7 @@ class NotThisMethod(Exception):
 
 
 def register_vcs_handler(vcs, method):  # decorator
-    """Decorator to mark a method as the handler for a particular VCS."""
+    """Mark a method as the handler for a particular VCS (decorator)."""
 
     def decorate(f):
         """Store f in HANDLERS[vcs][method]."""

coco/blocklist.py

@@ -24,7 +24,6 @@ class Blocklist:
     """
 
     def __init__(self, hosts, path: os.PathLike):
-
         # Initialise persistent storage
         self._state = PersistentState(path)
         if self._state.state is None:
@@ -177,7 +176,6 @@ def _check_hosts(self, hosts: List[Host]) -> Tuple[List[Host], List[bool]]:
         """
 
         def _check_host(host):
-
             if not isinstance(host, Host):
                 host = Host(host)
 

coco/check.py

@@ -116,7 +116,11 @@ def _save_reply(self, reply):
         for r in reply.values():
             if isinstance(r, dict):
                 merged.update(r)
-        self.state.write(self.save_to_state, merged)
+        if isinstance(self.save_to_state, str):
+            self.state.write(self.save_to_state, merged)
+        else:
+            for key, targ in self.save_to_state.items():
+                self.state.write(key, merged[targ])
 
 
 class ReplyCheck(Check):
@@ -413,6 +417,12 @@ async def run(self, result: Result):
                         failed_hosts.add(host)
                         result.report_failure(self._name, host, "missing", name)
                     continue
+                if result_ == "Timeout":
+                    for name in self.state_paths.keys():
+                        logger.debug(f"/{self._name}: Reply timed out for {host}.")
+                        failed_hosts.add(host)
+                        result.report_failure(self._name, host, "timeout", name)
+                    continue
                 for name, value in result_.items():
                     if name not in self.state_paths:
                         logger.debug(

coco/config.py

@@ -160,7 +160,6 @@ def load_config(path=None):
     any_exist = False
 
     for cfile in config_files:
-
         # Expand the configuration file path
         absfile = Path(cfile).expanduser().resolve()
 

coco/core.py

@@ -15,7 +15,13 @@
 
 import json
 import redis
-import aioredis
+
+import sys
+
+if sys.version_info.minor <= 10:
+    import aioredis
+else:
+    from redis import asyncio as aioredis
 
 from sanic import Sanic, response
 
@@ -29,17 +35,24 @@
     Endpoint,
     LocalEndpoint,
 )
+from .result import Result
 from . import worker, __version__, wait
 from .state import State
 from .exceptions import ConfigError, InternalError
 from .util import Host, str2total_seconds
 from . import slack
 from . import config
 
+Sanic.START_METHOD_SET = True
+Sanic.start_method = "fork"
+
 logger = logging.getLogger(__name__)
 
 # This should be a no-op on Linux but is required on MacOS for coco to run
-set_start_method("fork")
+try:
+    set_start_method("fork", force=True)
+except RuntimeError:
+    pass
 
 
 class Core:
@@ -166,7 +179,6 @@ def __del__(self):
                 logger.error(
                     f"Failed sending shutdown command to worker (have to kill it): {type(e)}: {e}"
                 )
-                self._kill_worker()
             self._kill_worker()
 
     def _kill_worker(self):
@@ -254,7 +266,6 @@ def _config_slack_loggers(self):
         slack.set_token(self.config["slack_token"])
 
         for rule in self.config["slack_rules"]:
-
             logger_name = rule["logger"]
             channel = rule["channel"]
             level = rule.get("level", "INFO").upper()
@@ -292,7 +303,6 @@ def _register_config(self):
             logger.warning("Config registration DISABLED. This is only OK for testing.")
 
     def _load_config(self, config_path: os.PathLike):
-
         self.config = config.load_config(config_path)
 
         self.log_level = self.config["log_level"]
@@ -337,11 +347,9 @@ def _load_config(self, config_path: os.PathLike):
                 logger.error(f"Invalid slack rule {rdict}.")
 
     def _load_endpoints(self):
-
         self.endpoints = {}
 
         for conf in self.config["endpoints"]:
-
             name = conf["name"]
 
             # Create the endpoint object
@@ -371,13 +379,21 @@ def _local_endpoints(self):
             "reset-state": ("POST", self.state.reset_state),
             "save-state": ("POST", self.state.save_state),
             "load-state": ("POST", self.state.load_state),
+            "get-coco-config": ("GET", self._get_config),
             "wait": ("POST", wait.process_post),
         }
 
         for name, (type_, callable_) in endpoints.items():
             self.endpoints[name] = LocalEndpoint(name, type_, callable_)
             self.forwarder.add_endpoint(name, self.endpoints[name])
 
+    async def _get_config(self, _):
+        return Result(
+            "coco-config",
+            result={Host("coco"): (self.config, 200)},
+            type_="FULL",
+        )
+
     def _check_endpoint_links(self):
         def check(e):
             if e:
@@ -414,7 +430,7 @@ async def external_endpoint(self, request, endpoint):
         Core endpoint. Passes all endpoint calls on to redis and blocks until completion.
         """
         # create a unique name for this task: <process ID>-<POSIX timestamp>
-        now = time.time()
+        now = time.perf_counter()
         name = f"{os.getpid()}-{now}"
 
         async with self.redis_async.client() as ra_cli:

coco/endpoint.py

@@ -50,6 +50,7 @@ def __init__(self, name, conf, forwarder, state):
         self.type = conf.get("type", "GET")
         self.group = conf.get("group")
         self.callable = conf.get("callable", False)
+        self.report_latency = conf.get("report_latency", True)
         self.call_on_start = conf.get("call_on_start", False)
         self.forwarder = forwarder
         self.state = state
@@ -109,7 +110,6 @@ def __init__(self, name, conf, forwarder, state):
 
                 # If save_state is set, the configured values have to match.
                 if self.values:
-
                     # Check if endpoint value types match the associated part of the saved state
                     for key in self.values.keys():
                         try:
@@ -183,7 +183,7 @@ def __init__(self, name, conf, forwarder, state):
 
     def _load_internal_forward(self, dict_, list_):
         """
-        Load Forward's from the config dictionary, generate objects and place in list.
+        Load Forwards from the config dictionary, generate objects and place in list.
 
         Parameters
         ----------
@@ -298,11 +298,11 @@ def _load_checks(self, check_dict: Dict) -> List[Check]:
 
         save_to_state = check_dict.get("save_reply_to_state", None)
         if save_to_state:
-            if not isinstance(save_to_state, str):
+            if not isinstance(save_to_state, (str, dict)):
                 raise ConfigError(
                     f"'save_reply_to_state' in check for '{name}' in '{self.name}"
                     f".conf' is of type '{type(save_to_state).__name__}' "
-                    f"(expected str)."
+                    f"(expected str or dict)."
                 )
             logger.debug(
                 f"Endpoint {self.name} will save replies to state: {save_to_state}."
@@ -514,6 +514,8 @@ async def call(self, request, hosts=None, params=None):
             self.write_timestamp()
             self.logger.debug("Success!")
 
+        result.report_latency = self.report_latency
+
         return result
 
     def write_timestamp(self):

coco/request_forwarder.py

@@ -46,7 +46,6 @@ def _create_trace_callback(name):
     """
 
     async def _callback(session, context, params):  # pylint: disable=W0613
-
         if not hasattr(context, "event_status"):
             context.event_status = {}
             context.start_time = time.time()
@@ -62,7 +61,6 @@ def _trace_config(trace_all=False):
     If trace_all=False, only dump on exceptions, otherwise dump at the end of a request too.
     """
     if not hasattr(_trace_config, "obj"):
-
         _trace_config.obj = aiohttp.TraceConfig()
 
         # This is a big hack, but it finds the events by looking for the underlying
@@ -353,7 +351,7 @@ async def _request(self, session, method, host, endpoint, request, params, timeo
         """
         url = host.join_endpoint(endpoint)
         hostname, port = host.hostname, host.port
-        start_time = time.time()
+        start_time = time.perf_counter()
         status = "0"
         try:
             async with session.request(
@@ -364,20 +362,25 @@ async def _request(self, session, method, host, endpoint, request, params, timeo
                 timeout=aiohttp.ClientTimeout(timeout),
                 params=params,
             ) as response:
+                response_time = time.perf_counter() - start_time
                 try:
                     status = str(response.status)
                     return (
                         host,
-                        (await response.json(content_type=None), response.status),
+                        (
+                            await response.json(content_type=None),
+                            response.status,
+                            response_time,
+                        ),
                     )
                 except json.decoder.JSONDecodeError:
-                    return host, (await response.text(), response.status)
+                    return host, (await response.text(), response.status, response_time)
         except AsyncioTimeoutError:
-            return host, ("Timeout", 0)
+            return host, ("Timeout", 0, 0)
         except Exception as e:
-            return host, (str(e), 0)
+            return host, (str(e), 0, 0)
         finally:
-            response_time = time.time() - start_time
+            response_time = time.perf_counter() - start_time
             self.response_time.labels(
                 endpoint=endpoint, host=hostname, port=port
             ).observe(response_time)