From a4f531ba3e0de0f1f483bbded7d03a4d4cab9854 Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Wed, 30 Jul 2025 22:44:24 +0100 Subject: [PATCH] Logging strategy updates --- .../trustgraph/gateway/config/receiver.py | 19 +++++++++++-------- .../dispatch/document_embeddings_export.py | 6 +++++- .../dispatch/entity_contexts_export.py | 6 +++++- .../dispatch/graph_embeddings_export.py | 6 +++++- .../trustgraph/gateway/dispatch/mux.py | 10 +++++++--- .../trustgraph/gateway/dispatch/requestor.py | 2 +- .../gateway/dispatch/triples_export.py | 6 +++++- .../gateway/endpoint/constant_endpoint.py | 2 +- .../trustgraph/gateway/endpoint/metrics.py | 2 +- .../gateway/endpoint/stream_endpoint.py | 2 +- .../gateway/endpoint/variable_endpoint.py | 2 +- 11 files changed, 43 insertions(+), 20 deletions(-) diff --git a/trustgraph-flow/trustgraph/gateway/config/receiver.py b/trustgraph-flow/trustgraph/gateway/config/receiver.py index 63800a41..0427e236 100755 --- a/trustgraph-flow/trustgraph/gateway/config/receiver.py +++ b/trustgraph-flow/trustgraph/gateway/config/receiver.py @@ -18,6 +18,9 @@ import logging import os import base64 import uuid + +# Module logger +logger = logging.getLogger(__name__) import json import pulsar @@ -48,7 +51,7 @@ class ConfigReceiver: v = msg.value() - print(f"Config version", v.version) + logger.info(f"Config version: {v.version}") if "flows" in v.config: @@ -68,29 +71,29 @@ class ConfigReceiver: del self.flows[k] except Exception as e: - print(f"Exception: {e}", flush=True) + logger.error(f"Config processing exception: {e}", exc_info=True) async def start_flow(self, id, flow): - print("Start flow", id) + logger.info(f"Starting flow: {id}") for handler in self.flow_handlers: try: await handler.start_flow(id, flow) except Exception as e: - print(f"Exception: {e}", flush=True) + logger.error(f"Config processing exception: {e}", exc_info=True) async def stop_flow(self, id, flow): - print("Stop flow", id) + logger.info(f"Stopping flow: {id}") for handler in self.flow_handlers: try: await handler.stop_flow(id, flow) except Exception as e: - print(f"Exception: {e}", flush=True) + logger.error(f"Config processing exception: {e}", exc_info=True) async def config_loader(self): @@ -111,9 +114,9 @@ class ConfigReceiver: await self.config_cons.start() - print("Waiting...") + logger.debug("Waiting for config updates...") - print("Config consumer done. :/") + logger.info("Config consumer finished") async def start(self): diff --git a/trustgraph-flow/trustgraph/gateway/dispatch/document_embeddings_export.py b/trustgraph-flow/trustgraph/gateway/dispatch/document_embeddings_export.py index 2587132d..1c65e8b3 100644 --- a/trustgraph-flow/trustgraph/gateway/dispatch/document_embeddings_export.py +++ b/trustgraph-flow/trustgraph/gateway/dispatch/document_embeddings_export.py @@ -2,12 +2,16 @@ import asyncio import queue import uuid +import logging from ... schema import DocumentEmbeddings from ... base import Subscriber from . serialize import serialize_document_embeddings +# Module logger +logger = logging.getLogger(__name__) + class DocumentEmbeddingsExport: def __init__( @@ -55,7 +59,7 @@ class DocumentEmbeddingsExport: continue except Exception as e: - print(f"Exception: {str(e)}", flush=True) + logger.error(f"Exception: {str(e)}", exc_info=True) break await subs.unsubscribe_all(id) diff --git a/trustgraph-flow/trustgraph/gateway/dispatch/entity_contexts_export.py b/trustgraph-flow/trustgraph/gateway/dispatch/entity_contexts_export.py index e388003b..9585c1d0 100644 --- a/trustgraph-flow/trustgraph/gateway/dispatch/entity_contexts_export.py +++ b/trustgraph-flow/trustgraph/gateway/dispatch/entity_contexts_export.py @@ -2,12 +2,16 @@ import asyncio import queue import uuid +import logging from ... schema import EntityContexts from ... base import Subscriber from . serialize import serialize_entity_contexts +# Module logger +logger = logging.getLogger(__name__) + class EntityContextsExport: def __init__( @@ -55,7 +59,7 @@ class EntityContextsExport: continue except Exception as e: - print(f"Exception: {str(e)}", flush=True) + logger.error(f"Exception: {str(e)}", exc_info=True) break await subs.unsubscribe_all(id) diff --git a/trustgraph-flow/trustgraph/gateway/dispatch/graph_embeddings_export.py b/trustgraph-flow/trustgraph/gateway/dispatch/graph_embeddings_export.py index 07f72550..44c70dfd 100644 --- a/trustgraph-flow/trustgraph/gateway/dispatch/graph_embeddings_export.py +++ b/trustgraph-flow/trustgraph/gateway/dispatch/graph_embeddings_export.py @@ -2,12 +2,16 @@ import asyncio import queue import uuid +import logging from ... schema import GraphEmbeddings from ... base import Subscriber from . serialize import serialize_graph_embeddings +# Module logger +logger = logging.getLogger(__name__) + class GraphEmbeddingsExport: def __init__( @@ -55,7 +59,7 @@ class GraphEmbeddingsExport: continue except Exception as e: - print(f"Exception: {str(e)}", flush=True) + logger.error(f"Exception: {str(e)}", exc_info=True) break await subs.unsubscribe_all(id) diff --git a/trustgraph-flow/trustgraph/gateway/dispatch/mux.py b/trustgraph-flow/trustgraph/gateway/dispatch/mux.py index 463d6dc9..afce6b75 100644 --- a/trustgraph-flow/trustgraph/gateway/dispatch/mux.py +++ b/trustgraph-flow/trustgraph/gateway/dispatch/mux.py @@ -2,6 +2,10 @@ import asyncio import queue import uuid +import logging + +# Module logger +logger = logging.getLogger(__name__) MAX_OUTSTANDING_REQUESTS = 15 WORKER_CLOSE_WAIT = 0.01 @@ -46,7 +50,7 @@ class Mux: )) except Exception as e: - print("receive exception:", str(e), flush=True) + logger.error(f"Receive exception: {str(e)}", exc_info=True) await self.ws.send_json({"error": str(e)}) async def maybe_tidy_workers(self, workers): @@ -138,7 +142,7 @@ class Mux: except Exception as e: # This is an internal working error, may not be recoverable - print("run prepare exception:", e) + logger.error(f"Run prepare exception: {e}", exc_info=True) await self.ws.send_json({"id": id, "error": str(e)}) self.running.stop() @@ -155,7 +159,7 @@ class Mux: ) except Exception as e: - print("Exception2:", e) + logger.error(f"Exception in mux: {e}", exc_info=True) await self.ws.send_json({"error": str(e)}) self.running.stop() diff --git a/trustgraph-flow/trustgraph/gateway/dispatch/requestor.py b/trustgraph-flow/trustgraph/gateway/dispatch/requestor.py index 35c41c8f..1acac5e5 100644 --- a/trustgraph-flow/trustgraph/gateway/dispatch/requestor.py +++ b/trustgraph-flow/trustgraph/gateway/dispatch/requestor.py @@ -68,7 +68,7 @@ class ServiceRequestor: q.get(), timeout=self.timeout ) except Exception as e: - print("Exception", e) + logger.error(f"Request timeout exception: {e}", exc_info=True) raise RuntimeError("Timeout") if resp.error: diff --git a/trustgraph-flow/trustgraph/gateway/dispatch/triples_export.py b/trustgraph-flow/trustgraph/gateway/dispatch/triples_export.py index d065550e..2847c182 100644 --- a/trustgraph-flow/trustgraph/gateway/dispatch/triples_export.py +++ b/trustgraph-flow/trustgraph/gateway/dispatch/triples_export.py @@ -2,12 +2,16 @@ import asyncio import queue import uuid +import logging from ... schema import Triples from ... base import Subscriber from . serialize import serialize_triples +# Module logger +logger = logging.getLogger(__name__) + class TriplesExport: def __init__( @@ -55,7 +59,7 @@ class TriplesExport: continue except Exception as e: - print(f"Exception: {str(e)}", flush=True) + logger.error(f"Exception: {str(e)}", exc_info=True) break await subs.unsubscribe_all(id) diff --git a/trustgraph-flow/trustgraph/gateway/endpoint/constant_endpoint.py b/trustgraph-flow/trustgraph/gateway/endpoint/constant_endpoint.py index 1e1d9d28..58ba1738 100644 --- a/trustgraph-flow/trustgraph/gateway/endpoint/constant_endpoint.py +++ b/trustgraph-flow/trustgraph/gateway/endpoint/constant_endpoint.py @@ -29,7 +29,7 @@ class ConstantEndpoint: async def handle(self, request): - print(request.path, "...") + logger.debug(f"Processing request: {request.path}") try: ht = request.headers["Authorization"] diff --git a/trustgraph-flow/trustgraph/gateway/endpoint/metrics.py b/trustgraph-flow/trustgraph/gateway/endpoint/metrics.py index d8a1ef62..d17d111b 100644 --- a/trustgraph-flow/trustgraph/gateway/endpoint/metrics.py +++ b/trustgraph-flow/trustgraph/gateway/endpoint/metrics.py @@ -33,7 +33,7 @@ class MetricsEndpoint: async def handle(self, request): - print(request.path, "...") + logger.debug(f"Processing metrics request: {request.path}") try: ht = request.headers["Authorization"] diff --git a/trustgraph-flow/trustgraph/gateway/endpoint/stream_endpoint.py b/trustgraph-flow/trustgraph/gateway/endpoint/stream_endpoint.py index 649c043e..38d8846f 100644 --- a/trustgraph-flow/trustgraph/gateway/endpoint/stream_endpoint.py +++ b/trustgraph-flow/trustgraph/gateway/endpoint/stream_endpoint.py @@ -36,7 +36,7 @@ class StreamEndpoint: async def handle(self, request): - print(request.path, "...") + logger.debug(f"Processing request: {request.path}") try: ht = request.headers["Authorization"] diff --git a/trustgraph-flow/trustgraph/gateway/endpoint/variable_endpoint.py b/trustgraph-flow/trustgraph/gateway/endpoint/variable_endpoint.py index ae0ae8fb..608de71b 100644 --- a/trustgraph-flow/trustgraph/gateway/endpoint/variable_endpoint.py +++ b/trustgraph-flow/trustgraph/gateway/endpoint/variable_endpoint.py @@ -28,7 +28,7 @@ class VariableEndpoint: async def handle(self, request): - print(request.path, "...") + logger.debug(f"Processing request: {request.path}") try: ht = request.headers["Authorization"]