From 22f3f345112b161f79aa635e9cb794cc547920bd Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Wed, 30 Jul 2025 22:57:41 +0100 Subject: [PATCH] Logging strategy updates --- trustgraph-base/trustgraph/base/consumer.py | 27 ++++++++++--------- .../trustgraph/base/embeddings_service.py | 12 ++++++--- .../trustgraph/base/llm_service.py | 8 ++++-- trustgraph-base/trustgraph/base/subscriber.py | 15 ++++++----- .../trustgraph/base/tool_service.py | 8 ++++-- .../trustgraph/base/triples_query_service.py | 5 ++++ .../trustgraph/base/triples_store_service.py | 7 ++++- 7 files changed, 54 insertions(+), 28 deletions(-) diff --git a/trustgraph-base/trustgraph/base/consumer.py b/trustgraph-base/trustgraph/base/consumer.py index 8b7b2b0d..43b4bc51 100644 --- a/trustgraph-base/trustgraph/base/consumer.py +++ b/trustgraph-base/trustgraph/base/consumer.py @@ -14,9 +14,13 @@ import pulsar import _pulsar import asyncio import time +import logging from .. exceptions import TooManyRequests +# Module logger +logger = logging.getLogger(__name__) + class Consumer: def __init__( @@ -90,7 +94,7 @@ class Consumer: try: - print(self.topic, "subscribing...", flush=True) + logger.info(f"Subscribing to topic: {self.topic}") if self.start_of_messages: pos = pulsar.InitialPosition.Earliest @@ -108,21 +112,18 @@ class Consumer: except Exception as e: - print("consumer subs Exception:", e, flush=True) + logger.error(f"Consumer subscription exception: {e}", exc_info=True) await asyncio.sleep(self.reconnect_time) continue - print(self.topic, "subscribed", flush=True) + logger.info(f"Successfully subscribed to topic: {self.topic}") if self.metrics: self.metrics.state("running") try: - print( - "Starting", self.concurrency, "receiver threads", - flush=True - ) + logger.info(f"Starting {self.concurrency} receiver threads") async with asyncio.TaskGroup() as tg: @@ -138,7 +139,7 @@ class Consumer: except Exception as e: - print("consumer loop exception:", e, flush=True) + logger.error(f"Consumer loop exception: {e}", exc_info=True) self.consumer.unsubscribe() self.consumer.close() self.consumer = None @@ -174,7 +175,7 @@ class Consumer: if time.time() > expiry: - print("Gave up waiting for rate-limit retry", flush=True) + logger.warning("Gave up waiting for rate-limit retry") # Message failed to be processed, this causes it to # be retried @@ -188,7 +189,7 @@ class Consumer: try: - print("Handle...", flush=True) + logger.debug("Processing message...") if self.metrics: @@ -198,7 +199,7 @@ class Consumer: else: await self.handler(msg, self, self.flow) - print("Handled.", flush=True) + logger.debug("Message processed successfully") # Acknowledge successful processing of the message self.consumer.acknowledge(msg) @@ -211,7 +212,7 @@ class Consumer: except TooManyRequests: - print("TooManyRequests: will retry...", flush=True) + logger.warning("Rate limit exceeded, will retry...") if self.metrics: self.metrics.rate_limit() @@ -224,7 +225,7 @@ class Consumer: except Exception as e: - print("consume exception:", e, flush=True) + logger.error(f"Message processing exception: {e}", exc_info=True) # Message failed to be processed, this causes it to # be retried diff --git a/trustgraph-base/trustgraph/base/embeddings_service.py b/trustgraph-base/trustgraph/base/embeddings_service.py index c0dd3978..556d32ff 100644 --- a/trustgraph-base/trustgraph/base/embeddings_service.py +++ b/trustgraph-base/trustgraph/base/embeddings_service.py @@ -4,12 +4,16 @@ Embeddings resolution base class """ import time +import logging from prometheus_client import Histogram from .. schema import EmbeddingsRequest, EmbeddingsResponse, Error from .. exceptions import TooManyRequests from .. base import FlowProcessor, ConsumerSpec, ProducerSpec +# Module logger +logger = logging.getLogger(__name__) + default_ident = "embeddings" default_concurrency = 1 @@ -51,7 +55,7 @@ class EmbeddingsService(FlowProcessor): id = msg.properties()["id"] - print("Handling request", id, "...", flush=True) + logger.debug(f"Handling embeddings request {id}...") vectors = await self.on_embeddings(request.text) @@ -63,7 +67,7 @@ class EmbeddingsService(FlowProcessor): properties={"id": id} ) - print("Handled.", flush=True) + logger.debug("Embeddings request handled successfully") except TooManyRequests as e: raise e @@ -72,9 +76,9 @@ class EmbeddingsService(FlowProcessor): # Apart from rate limits, treat all exceptions as unrecoverable - print(f"Exception: {e}", flush=True) + logger.error(f"Exception in embeddings service: {e}", exc_info=True) - print("Send error response...", flush=True) + logger.info("Sending error response...") await flow.producer["response"].send( EmbeddingsResponse( diff --git a/trustgraph-base/trustgraph/base/llm_service.py b/trustgraph-base/trustgraph/base/llm_service.py index fddbdf3e..37b0e1c2 100644 --- a/trustgraph-base/trustgraph/base/llm_service.py +++ b/trustgraph-base/trustgraph/base/llm_service.py @@ -4,12 +4,16 @@ LLM text completion base class """ import time +import logging from prometheus_client import Histogram from .. schema import TextCompletionRequest, TextCompletionResponse, Error from .. exceptions import TooManyRequests from .. base import FlowProcessor, ConsumerSpec, ProducerSpec +# Module logger +logger = logging.getLogger(__name__) + default_ident = "text-completion" default_concurrency = 1 @@ -103,9 +107,9 @@ class LlmService(FlowProcessor): # Apart from rate limits, treat all exceptions as unrecoverable - print(f"Exception: {e}") + logger.error(f"LLM service exception: {e}", exc_info=True) - print("Send error response...", flush=True) + logger.debug("Sending error response...") await flow.producer["response"].send( TextCompletionResponse( diff --git a/trustgraph-base/trustgraph/base/subscriber.py b/trustgraph-base/trustgraph/base/subscriber.py index 6e79adab..7b5fa6b5 100644 --- a/trustgraph-base/trustgraph/base/subscriber.py +++ b/trustgraph-base/trustgraph/base/subscriber.py @@ -7,6 +7,10 @@ from pulsar.schema import JsonSchema import asyncio import _pulsar import time +import logging + +# Module logger +logger = logging.getLogger(__name__) class Subscriber: @@ -66,7 +70,7 @@ class Subscriber: if self.metrics: self.metrics.state("running") - print("Subscriber running...", flush=True) + logger.info("Subscriber running...") while self.running: @@ -78,8 +82,7 @@ class Subscriber: except _pulsar.Timeout: continue except Exception as e: - print("Exception:", e, flush=True) - print(type(e)) + logger.error(f"Exception in subscriber receive: {e}", exc_info=True) raise e if self.metrics: @@ -110,7 +113,7 @@ class Subscriber: except Exception as e: self.metrics.dropped() - print("Q Put:", e, flush=True) + logger.warning(f"Failed to put message in queue: {e}") for q in self.full.values(): try: @@ -121,10 +124,10 @@ class Subscriber: ) except Exception as e: self.metrics.dropped() - print("Q Put:", e, flush=True) + logger.warning(f"Failed to put message in full queue: {e}") except Exception as e: - print("Subscriber exception:", e, flush=True) + logger.error(f"Subscriber exception: {e}", exc_info=True) finally: diff --git a/trustgraph-base/trustgraph/base/tool_service.py b/trustgraph-base/trustgraph/base/tool_service.py index 4f63bc53..f6924d52 100644 --- a/trustgraph-base/trustgraph/base/tool_service.py +++ b/trustgraph-base/trustgraph/base/tool_service.py @@ -4,12 +4,16 @@ Tool invocation base class """ import json +import logging from prometheus_client import Counter from .. schema import ToolRequest, ToolResponse, Error from .. exceptions import TooManyRequests from .. base import FlowProcessor, ConsumerSpec, ProducerSpec +# Module logger +logger = logging.getLogger(__name__) + default_concurrency = 1 class ToolService(FlowProcessor): @@ -91,9 +95,9 @@ class ToolService(FlowProcessor): # Apart from rate limits, treat all exceptions as unrecoverable - print(f"Exception: {e}") + logger.error(f"Exception in tool service: {e}", exc_info=True) - print("Send error response...", flush=True) + logger.info("Sending error response...") await flow.producer["response"].send( ToolResponse( diff --git a/trustgraph-base/trustgraph/base/triples_query_service.py b/trustgraph-base/trustgraph/base/triples_query_service.py index 37acc622..880c5a37 100644 --- a/trustgraph-base/trustgraph/base/triples_query_service.py +++ b/trustgraph-base/trustgraph/base/triples_query_service.py @@ -4,6 +4,8 @@ Triples query service. Input is a (s, p, o) triple, some values may be null. Output is a list of triples. """ +import logging + from .. schema import TriplesQueryRequest, TriplesQueryResponse, Error from .. schema import Value, Triple @@ -11,6 +13,9 @@ from . flow_processor import FlowProcessor from . consumer_spec import ConsumerSpec from . producer_spec import ProducerSpec +# Module logger +logger = logging.getLogger(__name__) + default_ident = "triples-query" class TriplesQueryService(FlowProcessor): diff --git a/trustgraph-base/trustgraph/base/triples_store_service.py b/trustgraph-base/trustgraph/base/triples_store_service.py index c33c2801..ac6e2298 100644 --- a/trustgraph-base/trustgraph/base/triples_store_service.py +++ b/trustgraph-base/trustgraph/base/triples_store_service.py @@ -3,10 +3,15 @@ Triples store base class """ +import logging + from .. schema import Triples from .. base import FlowProcessor, ConsumerSpec from .. exceptions import TooManyRequests +# Module logger +logger = logging.getLogger(__name__) + default_ident = "triples-write" class TriplesStoreService(FlowProcessor): @@ -38,7 +43,7 @@ class TriplesStoreService(FlowProcessor): except Exception as e: - print(f"Exception: {e}") + logger.error(f"Exception in triples store service: {e}", exc_info=True) raise e @staticmethod