async & flow donw

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

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)
 

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)