From 44b1d7e5080e4b093dcc82a8a86e6eca9acfdbdc Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Wed, 16 Apr 2025 21:34:39 +0100 Subject: [PATCH] async & flow donw --- Makefile | 4 +++ .../trustgraph/base/async_processor.py | 22 +++++++--------- .../trustgraph/base/flow_processor.py | 26 +++++++------------ trustgraph-base/trustgraph/base/metrics.py | 8 +++--- .../trustgraph/chunking/recursive/chunker.py | 2 +- .../trustgraph/chunking/token/chunker.py | 2 +- .../trustgraph/config/service/service.py | 17 +++--------- .../trustgraph/decoding/pdf/pdf_decoder.py | 2 +- 8 files changed, 33 insertions(+), 50 deletions(-) diff --git a/Makefile b/Makefile index a449d36b..1899e602 100644 --- a/Makefile +++ b/Makefile @@ -61,6 +61,10 @@ container: update-package-versions -t ${CONTAINER_BASE}/trustgraph-ocr:${VERSION} . some-containers: + ${DOCKER} build -f containers/Containerfile.base \ + -t ${CONTAINER_BASE}/trustgraph-base:${VERSION} . + ${DOCKER} build -f containers/Containerfile.flow \ + -t ${CONTAINER_BASE}/trustgraph-flow:${VERSION} . ${DOCKER} build -f containers/Containerfile.vertexai \ -t ${CONTAINER_BASE}/trustgraph-vertexai:${VERSION} . diff --git a/trustgraph-base/trustgraph/base/async_processor.py b/trustgraph-base/trustgraph/base/async_processor.py index 59c84a10..a4f4312a 100644 --- a/trustgraph-base/trustgraph/base/async_processor.py +++ b/trustgraph-base/trustgraph/base/async_processor.py @@ -17,6 +17,7 @@ from .. exceptions import TooManyRequests from . pubsub import PulsarClient from . producer import Producer from . consumer import Consumer +from . metrics import ProcessorMetrics default_config_queue = config_push_queue @@ -31,24 +32,19 @@ class AsyncProcessor: # Register a pulsar client self.client = PulsarClient(**params) + # Initialise metrics, records the parameters + ProcessorMetrics(id=self.id).info({ + k: str(params[k]) + for k in params + if k != "id" + }) + # The processor runs all activity in a taskgroup, it's mandatory # that this is provded self.taskgroup = params.get("taskgroup") if self.taskgroup is None: raise RuntimeError("Essential taskgroup missing") - # Pubsub parameters passed in - if not hasattr(__class__, "params_metric"): - __class__.params_metric = Info( - 'params', 'Parameters configuration' - ) - - # Record metrics for the processor - __class__.params_metric.info({ - k: str(params[k]) - for k in params - }) - # Get the configuration topic self.config_push_queue = params.get( "config_push_queue", default_config_queue @@ -161,7 +157,7 @@ class AsyncProcessor: # as a paramter. A processor identity ident is used as # - subscriber name # - an identifier for flow configuration - p = cls(**args | { "taskgroup": tg, "id", ident }) + p = cls(**args | { "taskgroup": tg, "id": ident }) # Start the processor await p.start() diff --git a/trustgraph-base/trustgraph/base/flow_processor.py b/trustgraph-base/trustgraph/base/flow_processor.py index c1a47119..25d9985f 100644 --- a/trustgraph-base/trustgraph/base/flow_processor.py +++ b/trustgraph-base/trustgraph/base/flow_processor.py @@ -12,8 +12,7 @@ from .. schema import config_request_queue, config_response_queue from .. schema import config_push_queue from .. log_level import LogLevel from .. base import AsyncProcessor, Consumer, Producer - -from .. base import ProcessorMetrics, ConsumerMetrics, ProducerMetrics +from . metrics import ConsumerMetrics, ProducerMetrics class Flow: def __init__(self, id, flow, processor, defn): @@ -30,15 +29,15 @@ class Flow: if not hasattr(self, name): setattr(self, name, defn[name]) - for spec in self.producer_spec: + for spec in processor.producer_spec: name, schema = spec producer_metrics = ProducerMetrics( - self.id, f"{flow}-{name}" + id, f"{flow}-{name}" ) - producer = self.publish( + producer = processor.publish( queue = defn[name], schema = schema, metrics = producer_metrics, @@ -49,18 +48,18 @@ class Flow: if not hasattr(self, name): setattr(self, name, producer) - for spec in self.consumer_spec: + for spec in processor.consumer_spec: name, schema, handler = spec consumer_metrics = ConsumerMetrics( - self.id, f"{flow}-{name}" + id, f"{flow}-{name}" ) - consumer = self.subscribe( - flow = flow_obj, + consumer = processor.subscribe( + flow = self, queue = defn[name], - subscriber = self.id, + subscriber = id, schema = schema, handler = handler, metrics = consumer_metrics, @@ -72,12 +71,10 @@ class Flow: consumer.name = name consumer.flow = self - await consumer.start() - self.consumer[name] = consumer if not hasattr(self, name): - setattr(flow_obj, name, consumer) + setattr(self, name, consumer) async def start(self): for c in self.consumer.values(): @@ -96,9 +93,6 @@ class FlowProcessor(AsyncProcessor): # Initialise base class super(FlowProcessor, self).__init__(**params) - # Initialise metrics, records the parameters - ProcessorMetrics(id=self.id).info(params) - # Register configuration handler self.register_config_handler(self.on_configuration) diff --git a/trustgraph-base/trustgraph/base/metrics.py b/trustgraph-base/trustgraph/base/metrics.py index 6088c123..5d87849f 100644 --- a/trustgraph-base/trustgraph/base/metrics.py +++ b/trustgraph-base/trustgraph/base/metrics.py @@ -71,12 +71,12 @@ class ProcessorMetrics: self.id = id - if not hasattr(__class__, "pubsub_metric"): - __class__.pubsub_metric = Info( - 'pubsub', 'Pub/sub configuration', + if not hasattr(__class__, "processor_metric"): + __class__.processor_metric = Info( + 'processor', 'Processor configuration', ["id"] ) def info(self, info): - __class__.pubsub_metric.labels(id=self.id).info(info) + __class__.processor_metric.labels(id=self.id).info(info) diff --git a/trustgraph-flow/trustgraph/chunking/recursive/chunker.py b/trustgraph-flow/trustgraph/chunking/recursive/chunker.py index 1106f02f..d03af3b1 100755 --- a/trustgraph-flow/trustgraph/chunking/recursive/chunker.py +++ b/trustgraph-flow/trustgraph/chunking/recursive/chunker.py @@ -101,5 +101,5 @@ class Processor(FlowProcessor): def run(): - Processor.launch(module, __doc__) + Processor.launch(default_ident, __doc__) diff --git a/trustgraph-flow/trustgraph/chunking/token/chunker.py b/trustgraph-flow/trustgraph/chunking/token/chunker.py index be2799df..76e49e87 100755 --- a/trustgraph-flow/trustgraph/chunking/token/chunker.py +++ b/trustgraph-flow/trustgraph/chunking/token/chunker.py @@ -100,5 +100,5 @@ class Processor(FlowProcessor): def run(): - Processor.launch(module, __doc__) + Processor.launch(default_ident, __doc__) diff --git a/trustgraph-flow/trustgraph/config/service/service.py b/trustgraph-flow/trustgraph/config/service/service.py index c441e640..bd1b1cb4 100644 --- a/trustgraph-flow/trustgraph/config/service/service.py +++ b/trustgraph-flow/trustgraph/config/service/service.py @@ -15,7 +15,7 @@ from trustgraph.base import AsyncProcessor, Consumer, Producer from . config import Configuration from ... base import ProcessorMetrics, ConsumerMetrics, ProducerMetrics -module = "config-svc" +default_ident = "config-svc" default_request_queue = config_request_queue default_response_queue = config_response_queue @@ -28,23 +28,12 @@ class Processor(AsyncProcessor): request_queue = params.get("request_queue", default_request_queue) response_queue = params.get("response_queue", default_response_queue) push_queue = params.get("push_queue", default_push_queue) - subscriber = params.get("subscriber", default_subscriber) id = params.get("id") request_schema = ConfigRequest response_schema = ConfigResponse push_schema = ConfigResponse - ProcessorMetrics(id=id).info( - { - "subscriber": subscriber, - "request_queue": request_queue, - "request_schema": request_schema.__name__, - "response_queue": response_queue, - "response_schema": request_schema.__name__, - } - ) - super(Processor, self).__init__( **params | { "request_schema": request_schema.__name__, @@ -72,7 +61,7 @@ class Processor(AsyncProcessor): self.subs = self.subscribe( flow = None, queue = request_queue, - subscriber = subscriber, + subscriber = id, schema = request_schema, handler = self.on_message, metrics = self.request_metrics, @@ -154,5 +143,5 @@ class Processor(AsyncProcessor): def run(): - Processor.launch(module, __doc__) + Processor.launch(default_ident, __doc__) diff --git a/trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py b/trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py index ae2e3d43..72ebe3dc 100755 --- a/trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py +++ b/trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py @@ -77,5 +77,5 @@ class Processor(FlowProcessor): def run(): - Processor.launch(ident, __doc__) + Processor.launch(default_ident, __doc__)