From 9fdc408a9569822147379e2be175928234affb5c Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Mon, 14 Apr 2025 22:39:09 +0100 Subject: [PATCH] Metrics working --- trustgraph-base/trustgraph/base/__init__.py | 1 + .../trustgraph/base/async_processor.py | 19 +++-- trustgraph-base/trustgraph/base/consumer.py | 48 ++++++++--- trustgraph-base/trustgraph/base/metrics.py | 82 +++++++++++++++++++ trustgraph-base/trustgraph/base/producer.py | 11 +-- .../trustgraph/config/service/service.py | 37 ++++----- 6 files changed, 152 insertions(+), 46 deletions(-) create mode 100644 trustgraph-base/trustgraph/base/metrics.py diff --git a/trustgraph-base/trustgraph/base/__init__.py b/trustgraph-base/trustgraph/base/__init__.py index 02548d66..4bf074e4 100644 --- a/trustgraph-base/trustgraph/base/__init__.py +++ b/trustgraph-base/trustgraph/base/__init__.py @@ -5,4 +5,5 @@ from . consumer import Consumer from . producer import Producer from . publisher import Publisher from . subscriber import Subscriber +from . metrics import ProcessorMetrics, ConsumerMetrics, ProducerMetrics diff --git a/trustgraph-base/trustgraph/base/async_processor.py b/trustgraph-base/trustgraph/base/async_processor.py index a7d7f221..39875342 100644 --- a/trustgraph-base/trustgraph/base/async_processor.py +++ b/trustgraph-base/trustgraph/base/async_processor.py @@ -73,16 +73,25 @@ class AsyncProcessor: while True: await asyncio.sleep(2) - def subscribe(self, queue, subscriber, schema, handler): + def subscribe(self, queue, subscriber, schema, handler, metrics=None): return Consumer( - self.taskgroup, self.client, queue, subscriber, schema, handler + taskgroup = self.taskgroup, + client = self.client, + queue = queue, + subscriber = subscriber, + schema = schema, + handler = handler, + metrics = metrics, ) - def publish(self, queue, schema): + def publish(self, queue, schema, metrics=None): return Producer( - self.client, queue, schema + client = self.client, + queue = queue, + schema = schema, + metrics = metrics, ) def set_processor_state(self, flow, state): @@ -166,7 +175,7 @@ class AsyncProcessor: print("Exception group:") for se in e.exceptions: - print(" Type:", type(se)) + print(" Type:", type(se)) print(f" Exception: {se}") except Exception as e: diff --git a/trustgraph-base/trustgraph/base/consumer.py b/trustgraph-base/trustgraph/base/consumer.py index 7ea9ef49..80fb25a5 100644 --- a/trustgraph-base/trustgraph/base/consumer.py +++ b/trustgraph-base/trustgraph/base/consumer.py @@ -8,7 +8,8 @@ class Consumer: def __init__( self, taskgroup, client, queue, subscriber, schema, - handler + handler, rate_limit_retry_time = 10, + rate_limit_timeout = 7200, metrics = None ): self.taskgroup = taskgroup @@ -17,10 +18,14 @@ class Consumer: self.subscriber = subscriber self.schema = schema self.handler = handler + self.rate_limit_retry_time = rate_limit_retry_time + self.rate_limit_timeout = rate_limit_timeout self.running = True self.task = None + self.metrics = metrics + async def start(self): self.running = True @@ -29,17 +34,23 @@ class Consumer: self.queue, self.subscriber, self.schema ) + # Puts it in the stopped state, the run thread should set running + if self.metrics: + self.metrics.state("stopped") + self.task = self.taskgroup.create_task(self.run()) - async def run(self): + if self.metrics: + print("RUNNING") + self.metrics.state("running") + while self.running: msg = await asyncio.to_thread(self.consumer.receive) -# expiry = time.time() + self.rate_limit_timeout - expiry = time.time() + 10 + expiry = time.time() + self.rate_limit_timeout # This loop is for retry on rate-limit / resource limits while True: @@ -52,8 +63,8 @@ class Consumer: # be retried self.consumer.negative_acknowledge(msg) - # FIXME -# __class__.processing_metric.labels(status="error").inc() + if self.metrics: + self.metrics.process("error") # Break out of retry loop, processes next message break @@ -61,15 +72,22 @@ class Consumer: try: print("Handle...") - # FIXME -# with __class__.request_metric.time(): - await self.handler(msg, self.consumer) + + if self.metrics: + + with self.metrics.record_time(): + await self.handler(msg, self.consumer) + + else: + await self.handler(msg, self.consumer) + print("Handled.") # Acknowledge successful processing of the message self.consumer.acknowledge(msg) -# __class__.processing_metric.labels(status="success").inc() + if self.metrics: + self.metrics.process("success") # Break out of retry loop break @@ -79,10 +97,11 @@ class Consumer: print("TooManyRequests: will retry...", flush=True) # FIXME -# __class__.rate_limit_metric.inc() + if self.metrics: + self.metrics.rate_limit() # Sleep - time.sleep(self.rate_limit_retry) + await asyncio.sleep(self.rate_limit_retry) # Contine from retry loop, just causes a reprocessing continue @@ -95,8 +114,11 @@ class Consumer: # be retried self.consumer.negative_acknowledge(msg) -# __class__.processing_metric.labels(status="error").inc() + if self.metrics: + self.metrics.process("error") # Break out of retry loop, processes next message break + if self.metrics: + self.metrics.state("stopped") diff --git a/trustgraph-base/trustgraph/base/metrics.py b/trustgraph-base/trustgraph/base/metrics.py new file mode 100644 index 00000000..6088c123 --- /dev/null +++ b/trustgraph-base/trustgraph/base/metrics.py @@ -0,0 +1,82 @@ + +from prometheus_client import start_http_server, Info, Enum, Histogram +from prometheus_client import Counter + +class ConsumerMetrics: + + def __init__(self, id, flow=None): + + self.id = id + self.flow = flow + + if not hasattr(__class__, "state_metric"): + __class__.state_metric = Enum( + 'consumer_state', 'Consumer state', + ["id", "flow"], + states=['stopped', 'running'] + ) + if not hasattr(__class__, "request_metric"): + __class__.request_metric = Histogram( + 'request_latency', 'Request latency (seconds)', + ["id", "flow"], + ) + if not hasattr(__class__, "processing_metric"): + __class__.processing_metric = Counter( + 'processing_count', 'Processing count', + ["id", "flow", "status"] + ) + if not hasattr(__class__, "rate_limit_metric"): + __class__.rate_limit_metric = Counter( + 'rate_limit_count', 'Rate limit event count', + ["id", "flow"] + ) + + def process(self, status): + __class__.processing_metric.labels( + id=self.id, flow=self.flow, status=status + ).inc() + + def rate_limit(self): + __class__.rate_limit_metric.labels( + id=self.id, flow=self.flow + ).inc() + + def state(self, state): + __class__.state_metric.labels( + id=self.id, flow=self.flow + ).state(state) + + def record_time(self): + return __class__.request_metric.labels( + id=self.id, flow=self.flow + ).time() + +class ProducerMetrics: + def __init__(self, id, flow=None): + + self.id = id + self.flow = flow + + if not hasattr(__class__, "output_metric"): + __class__.output_metric = Counter( + 'output_count', 'Output items created', + ["id", "flow"] + ) + + def inc(self): + __class__.output_metric.labels(id=self.id, flow=self.flow).inc() + +class ProcessorMetrics: + def __init__(self, id): + + self.id = id + + if not hasattr(__class__, "pubsub_metric"): + __class__.pubsub_metric = Info( + 'pubsub', 'Pub/sub configuration', + ["id"] + ) + + def info(self, info): + __class__.pubsub_metric.labels(id=self.id).info(info) + diff --git a/trustgraph-base/trustgraph/base/producer.py b/trustgraph-base/trustgraph/base/producer.py index 9cb1197d..dbbf0237 100644 --- a/trustgraph-base/trustgraph/base/producer.py +++ b/trustgraph-base/trustgraph/base/producer.py @@ -1,18 +1,19 @@ class Producer: - def __init__(self, client, queue, schema): + def __init__(self, client, queue, schema, metrics=None): self.client = client self.queue = queue self.schema = schema - self.producer = None + self.producer = self.client.publish(self.queue, self.schema) + + self.metrics = metrics async def send(self, msg, properties={}): - if self.producer == None: - self.producer = self.client.publish(self.queue, self.schema) - + if self.metrics: + self.metrics.inc() self.producer.send(msg, properties) diff --git a/trustgraph-flow/trustgraph/config/service/service.py b/trustgraph-flow/trustgraph/config/service/service.py index 09c96e0f..e64c31b2 100644 --- a/trustgraph-flow/trustgraph/config/service/service.py +++ b/trustgraph-flow/trustgraph/config/service/service.py @@ -15,6 +15,7 @@ from trustgraph.log_level import LogLevel from trustgraph.base import AsyncProcessor, Consumer, Producer from . config import Configuration +from ... base import ProcessorMetrics, ConsumerMetrics, ProducerMetrics module = "config-svc" @@ -40,19 +41,13 @@ class Processor(AsyncProcessor): response_schema = ConfigResponse push_schema = ConfigResponse - self.set_processor_state(id, "starting") - - self.set_pubsub_info( - id + ProcessorMetrics(id=id).info( { - "id": id, "subscriber": subscriber, "request_queue": request_queue, "request_schema": request_schema.__name__, "response_queue": response_queue, "response_schema": request_schema.__name__, -# "rate_limit_retry": str(self.rate_limit_retry), -# "rate_limit_timeout": str(self.rate_limit_timeout), } ) @@ -64,25 +59,20 @@ class Processor(AsyncProcessor): } ) - if not hasattr(__class__, "request_metric"): - __class__.request_metric = Histogram( - 'request_latency', 'Request latency (seconds)' - ) - - if not hasattr(__class__, "processing_metric"): - __class__.processing_metric = Counter( - 'processing_count', 'Processing count', - ["status"] - ) + self.request_metrics = ConsumerMetrics(id + "-request") + self.response_metrics = ProducerMetrics(id + "-response") + self.push_metrics = ProducerMetrics(id + "-push") self.push_pub = self.publish( queue = push_queue, - schema = ConfigPush + schema = ConfigPush, + metrics = self.push_metrics, ) - self.out_pub = self.publish( + self.response_pub = self.publish( queue = response_queue, - schema = ConfigResponse + schema = ConfigResponse, + metrics = self.response_metrics, ) self.subs = self.subscribe( @@ -90,6 +80,7 @@ class Processor(AsyncProcessor): subscriber = subscriber, schema = request_schema, handler = self.on_message, + metrics = self.request_metrics, ) self.config = Configuration() @@ -130,7 +121,7 @@ class Processor(AsyncProcessor): resp = await self.config.handle(v) - await self.out_pub.send(resp, properties={"id": id}) + await self.response_pub.send(resp, properties={"id": id}) consumer.acknowledge(msg) @@ -144,7 +135,7 @@ class Processor(AsyncProcessor): text=None, ) - await self.out_pub.send(resp, properties={"id": id}) + await self.response_pub.send(resp, properties={"id": id}) consumer.acknowledge(msg) @@ -166,7 +157,7 @@ class Processor(AsyncProcessor): ) parser.add_argument( - '-P', '--push-queue', + '--push-queue', default=default_push_queue, help=f'Config push queue (default: {default_push_queue})' )