Logging strategy updates

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(
+                logger.info(f"Starting {self.concurrency} receiver threads")
-                    "Starting", self.concurrency, "receiver threads",
-                    flush=True
-                )
 
                 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

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(

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(

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)
+                        logger.error(f"Exception in subscriber receive: {e}", exc_info=True)
-                        print(type(e))
                         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:
 

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(

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

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