From 950c624013471eddef01e7e7b75987c210136d91 Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Wed, 23 Apr 2025 11:28:19 +0100 Subject: [PATCH] - Rename 'client' to 'pulsar_client' in async_processor. Less likely to be a name clash. - Turned on Metrics API in api-gateway. --- trustgraph-base/trustgraph/base/async_processor.py | 10 +++++----- trustgraph-base/trustgraph/base/consumer_spec.py | 2 +- trustgraph-base/trustgraph/base/producer_spec.py | 2 +- .../trustgraph/base/request_response_spec.py | 2 +- trustgraph-base/trustgraph/base/subscriber_spec.py | 2 +- trustgraph-flow/trustgraph/config/service/service.py | 10 +++++----- trustgraph-flow/trustgraph/gateway/service.py | 10 +++++----- 7 files changed, 19 insertions(+), 19 deletions(-) diff --git a/trustgraph-base/trustgraph/base/async_processor.py b/trustgraph-base/trustgraph/base/async_processor.py index 696c167f..7a942a15 100644 --- a/trustgraph-base/trustgraph/base/async_processor.py +++ b/trustgraph-base/trustgraph/base/async_processor.py @@ -30,7 +30,7 @@ class AsyncProcessor: self.id = params.get("id") # Register a pulsar client - self.pulsar_client = PulsarClient(**params) + self.pulsar_client_object = PulsarClient(**params) # Initialise metrics, records the parameters ProcessorMetrics(processor = self.id).info({ @@ -61,7 +61,7 @@ class AsyncProcessor: self.config_sub_task = Consumer( taskgroup = self.taskgroup, - client = self.client, + client = self.pulsar_client, subscriber = config_subscriber_id, flow = None, @@ -85,16 +85,16 @@ class AsyncProcessor: # This is called to stop all threads. An over-ride point for extra # functionality def stop(self): - self.client.close() + self.pulsar_client.close() self.running = False # Returns the pulsar host @property - def pulsar_host(self): return self.client.pulsar_host + def pulsar_host(self): return self.pulsar_client_object.pulsar_host # Returns the pulsar client @property - def client(self): return self.pulsar_client.client + def pulsar_client(self): return self.pulsar_client_object.client # Register a new event handler for configuration change def register_config_handler(self, handler): diff --git a/trustgraph-base/trustgraph/base/consumer_spec.py b/trustgraph-base/trustgraph/base/consumer_spec.py index c4a37ef6..21497dc5 100644 --- a/trustgraph-base/trustgraph/base/consumer_spec.py +++ b/trustgraph-base/trustgraph/base/consumer_spec.py @@ -18,7 +18,7 @@ class ConsumerSpec(Spec): consumer = Consumer( taskgroup = processor.taskgroup, flow = flow, - client = processor.client, + client = processor.pulsar_client, topic = definition[self.name], subscriber = processor.id + "--" + self.name, schema = self.schema, diff --git a/trustgraph-base/trustgraph/base/producer_spec.py b/trustgraph-base/trustgraph/base/producer_spec.py index 0549f949..9c8bbc6a 100644 --- a/trustgraph-base/trustgraph/base/producer_spec.py +++ b/trustgraph-base/trustgraph/base/producer_spec.py @@ -15,7 +15,7 @@ class ProducerSpec(Spec): ) producer = Producer( - client = processor.client, + client = processor.pulsar_client, topic = definition[self.name], schema = self.schema, metrics = producer_metrics, diff --git a/trustgraph-base/trustgraph/base/request_response_spec.py b/trustgraph-base/trustgraph/base/request_response_spec.py index 5fcd92b3..368b5e88 100644 --- a/trustgraph-base/trustgraph/base/request_response_spec.py +++ b/trustgraph-base/trustgraph/base/request_response_spec.py @@ -121,7 +121,7 @@ class RequestResponseSpec(Spec): ) rr = self.impl( - client = processor.client, + client = processor.pulsar_client, subscription = flow.id, consumer_name = flow.id, request_topic = definition[self.request_name], diff --git a/trustgraph-base/trustgraph/base/subscriber_spec.py b/trustgraph-base/trustgraph/base/subscriber_spec.py index 6c3d7c80..d0f24a07 100644 --- a/trustgraph-base/trustgraph/base/subscriber_spec.py +++ b/trustgraph-base/trustgraph/base/subscriber_spec.py @@ -17,7 +17,7 @@ class SubscriberSpec(Spec): ) subscriber = Subscriber( - client = processor.client, + client = processor.pulsar_client, topic = definition[self.name], subscription = flow.id, consumer_name = flow.id, diff --git a/trustgraph-flow/trustgraph/config/service/service.py b/trustgraph-flow/trustgraph/config/service/service.py index 972de45e..c0268389 100644 --- a/trustgraph-flow/trustgraph/config/service/service.py +++ b/trustgraph-flow/trustgraph/config/service/service.py @@ -87,7 +87,7 @@ class Processor(AsyncProcessor): self.config_request_consumer = Consumer( taskgroup = self.taskgroup, - client = self.client, + client = self.pulsar_client, flow = None, topic = config_request_queue, subscriber = id, @@ -97,14 +97,14 @@ class Processor(AsyncProcessor): ) self.config_response_producer = Producer( - client = self.client, + client = self.pulsar_client, topic = config_response_queue, schema = ConfigResponse, metrics = config_response_metrics, ) self.config_push_producer = Producer( - client = self.client, + client = self.pulsar_client, topic = config_push_queue, schema = ConfigPush, metrics = config_push_metrics, @@ -112,7 +112,7 @@ class Processor(AsyncProcessor): self.flow_request_consumer = Consumer( taskgroup = self.taskgroup, - client = self.client, + client = self.pulsar_client, flow = None, topic = flow_request_queue, subscriber = id, @@ -122,7 +122,7 @@ class Processor(AsyncProcessor): ) self.flow_response_producer = Producer( - client = self.client, + client = self.pulsar_client, topic = flow_response_queue, schema = FlowResponse, metrics = flow_response_metrics, diff --git a/trustgraph-flow/trustgraph/gateway/service.py b/trustgraph-flow/trustgraph/gateway/service.py index 1944dd9e..d7df3240 100755 --- a/trustgraph-flow/trustgraph/gateway/service.py +++ b/trustgraph-flow/trustgraph/gateway/service.py @@ -266,11 +266,11 @@ class Api: # auth = self.auth, # services = self.services, # ), - # MetricsEndpoint( - # endpoint_path = "/api/v1/metrics", - # prometheus_url = self.prometheus_url, - # auth = self.auth, - # ), + MetricsEndpoint( + endpoint_path = "/api/v1/metrics", + prometheus_url = self.prometheus_url, + auth = self.auth, + ), ] for ep in self.endpoints: