Gather BIT metrics [implementation]

octopoes/octopoes/config/settings.py

@@ -89,3 +89,4 @@ def settings_customise_sources(
 DEFAULT_LIMIT = 50
 DEFAULT_OFFSET = 0
 QUEUE_NAME_OCTOPOES: str = "octopoes"
+GATHER_BIT_METRICS: bool = False

octopoes/octopoes/core/app.py

@@ -2,7 +2,7 @@
 
 from amqp import AMQPError
 
-from octopoes.config.settings import QUEUE_NAME_OCTOPOES, Settings
+from octopoes.config.settings import GATHER_BIT_METRICS, QUEUE_NAME_OCTOPOES, Settings
 from octopoes.core.service import OctopoesService
 from octopoes.events.manager import EventManager, get_rabbit_channel
 from octopoes.repositories.ooi_repository import XTDBOOIRepository
@@ -39,6 +39,9 @@ def bootstrap_octopoes(settings: Settings, client: str, xtdb_session: XTDBSessio
     origin_param_repository = XTDBOriginParameterRepository(event_manager, xtdb_session)
     scan_profile_repository = XTDBScanProfileRepository(event_manager, xtdb_session)
 
-    octopoes = OctopoesService(ooi_repository, origin_repository, origin_param_repository, scan_profile_repository)
-
-    return octopoes
+    if GATHER_BIT_METRICS:
+        return OctopoesService(
+            ooi_repository, origin_repository, origin_param_repository, scan_profile_repository, xtdb_session
+        )
+    else:
+        return OctopoesService(ooi_repository, origin_repository, origin_param_repository, scan_profile_repository)

octopoes/octopoes/core/service.py

@@ -3,10 +3,12 @@
 from collections.abc import Callable, ValuesView
 from datetime import datetime, timezone
 from logging import getLogger
+from time import perf_counter
 from typing import overload
 
 from bits.definitions import get_bit_definitions
 from bits.runner import BitRunner
+from pydantic import TypeAdapter
 
 from octopoes.config.settings import (
     DEFAULT_LIMIT,
@@ -42,6 +44,7 @@
 from octopoes.repositories.origin_parameter_repository import OriginParameterRepository
 from octopoes.repositories.origin_repository import OriginRepository
 from octopoes.repositories.scan_profile_repository import ScanProfileRepository
+from octopoes.xtdb.client import Operation, OperationType, XTDBSession
 
 logger = getLogger(__name__)
 settings = Settings()
@@ -66,11 +69,13 @@ def __init__(
         origin_repository: OriginRepository,
         origin_parameter_repository: OriginParameterRepository,
         scan_profile_repository: ScanProfileRepository,
+        session: XTDBSession | None = None,
     ):
         self.ooi_repository = ooi_repository
         self.origin_repository = origin_repository
         self.origin_parameter_repository = origin_parameter_repository
         self.scan_profile_repository = scan_profile_repository
+        self.session = session
 
     @overload
     def _populate_scan_profiles(self, oois: ValuesView[OOI], valid_time: datetime) -> ValuesView[OOI]: ...
@@ -199,13 +204,29 @@ def _run_inference(self, origin: Origin, valid_time: datetime) -> None:
                 config = configs[-1].config
 
         try:
-            resulting_oois = BitRunner(bit_definition).run(source, list(parameters.values()), config=config)
+            if isinstance(self.session, XTDBSession):
+                start = perf_counter()
+                resulting_oois = BitRunner(bit_definition).run(source, list(parameters.values()), config=config)
+                stop = perf_counter()
+                metrics: dict[str, str] = {
+                    "bit": bit_definition.id,
+                    "config": json.dumps(config),
+                    "elapsed": str(stop - start),
+                    "parameters": TypeAdapter(list[OOI]).dump_json(list(parameters.values())).decode(),
+                    "source": source.model_dump_json(),
+                    "xt/id": "BIT_METRIC",
+                    "yield": TypeAdapter(list[OOI]).dump_json(resulting_oois).decode(),
+                }
+                ops: list[Operation] = [(OperationType.PUT, metrics, valid_time)]
+                self.session.client.submit_transaction(ops)
+                self.save_origin(origin, resulting_oois, valid_time)
+            else:
+                resulting_oois = BitRunner(bit_definition).run(source, list(parameters.values()), config=config)
+                self.save_origin(origin, resulting_oois, valid_time)
         except Exception as e:
             logger.exception("Error running inference", exc_info=e)
             return
 
-        self.save_origin(origin, resulting_oois, valid_time)
-
     @staticmethod
     def check_path_level(path_level: int | None, current_level: int):
         return path_level is not None and path_level >= current_level

octopoes/tools/analyze-bit-metric.py

@@ -0,0 +1,103 @@
+#!/usr/bin/env python
+
+import json
+import logging
+from datetime import datetime
+
+import click
+from xtdb_client import XTDBClient
+
+logger = logging.getLogger(__name__)
+
+
+class BitMetric:
+    def __init__(self, data):
+        date_format = "%Y-%m-%dT%H:%M:%SZ"
+        self.txTime: datetime = datetime.strptime(data["txTime"], date_format)
+        self.txId: int = int(data["txId"])
+        self.validTime: datetime = datetime.strptime(data["validTime"], date_format)
+        self.contentHash: str = data["contentHash"]
+        self.yld: dict[str, str] = json.loads(data["doc"]["yield"])
+        self.cfg: dict[str, str] = json.loads(data["doc"]["config"])
+        self.src: dict[str, str] = json.loads(data["doc"]["source"])
+        self.name: str = data["doc"]["bit"]
+        self.pms: list[dict[str, str]] = json.loads(data["doc"]["parameters"])
+        self.elapsed: list[dict[str, str]] = json.loads(data["doc"]["elapsed"])
+
+    def empty(self):
+        return len(self.yld) == 0
+
+    def __eq__(self, val):
+        return all([getattr(self, op) == getattr(val, op) for op in ["src", "cfg", "yld", "name", "pms"]])
+
+    def __hash__(self):
+        return hash(str(hash(self.txTime)) + self.contentHash)
+
+
+@click.group(
+    context_settings={
+        "help_option_names": ["-h", "--help"],
+        "max_content_width": 120,
+        "show_default": True,
+    }
+)
+@click.option("-n", "--node", default="0", help="XTDB node")
+@click.option(
+    "-u",
+    "--url",
+    default="http://localhost:3000",
+    help="XTDB server base url",
+)
+@click.option(
+    "-t",
+    "--timeout",
+    type=int,
+    default=5000,
+    help="XTDB request timeout (in ms)",
+)
+@click.option("-v", "--verbosity", count=True, help="Increase the verbosity level")
+@click.pass_context
+def cli(ctx: click.Context, url: str, node: str, timeout: int, verbosity: int):
+    verbosities = [logging.WARN, logging.INFO, logging.DEBUG]
+    try:
+        if verbosity:
+            logging.basicConfig(level=verbosities[verbosity - 1])
+    except IndexError:
+        raise click.UsageError("Invalid verbosity level (use -v, -vv, or -vvv)")
+
+    client = XTDBClient(url, node, timeout)
+    logger.info("Instantiated XTDB client with endpoint %s for node %s", url, node)
+
+    ctx.ensure_object(dict)
+    ctx.obj["client"] = client
+    ctx.obj["raw_bit_metrics"] = client.history("BIT_METRIC", False, True)
+    ctx.obj["bit_metrics"] = list(map(lambda x: BitMetric(x), ctx.obj["raw_bit_metrics"]))
+
+
+@cli.command(help="Returns the raw bit metric")
+@click.pass_context
+def raw(ctx: click.Context):
+    click.echo(ctx.obj["raw_bit_metrics"])
+
+
+@cli.command(help="Returns the parsed metric")
+@click.pass_context
+def parse(ctx: click.Context):
+    metrics = {
+        "total": len(ctx.obj["bit_metrics"]),
+        "total_elapsed": sum([bm.elapsed for bm in ctx.obj["bit_metrics"]]),
+        "empty": len([bm for bm in ctx.obj["bit_metrics"] if bm.empty()]),
+        "empty_elapsed": sum([bm.elapsed for bm in ctx.obj["bit_metrics"] if bm.empty()]),
+        "useful": len([bm for bm in ctx.obj["bit_metrics"] if not bm.empty()]),
+        "useful_elapsed": sum([bm.elapsed for bm in ctx.obj["bit_metrics"] if not bm.empty()]),
+        "futile_runs": {
+            bm.name: ctx.obj["bit_metrics"].count(bm) - 1
+            for bm in ctx.obj["bit_metrics"]
+            if ctx.obj["bit_metrics"].count(bm) > 1
+        },
+    }
+    click.echo(json.dumps(metrics))
+
+
+if __name__ == "__main__":
+    cli()