diff --git a/tests/unit/test_base/test_metrics.py b/tests/unit/test_base/test_metrics.py index 0496db20..dab9008b 100644 --- a/tests/unit/test_base/test_metrics.py +++ b/tests/unit/test_base/test_metrics.py @@ -73,8 +73,8 @@ def test_consumer_metrics_reuses_singletons_and_records_events(monkeypatch): monkeypatch.setattr(metrics, "Histogram", histogram_factory) monkeypatch.setattr(metrics, "Counter", counter_factory) - first = metrics.ConsumerMetrics("proc", "flow", "name") - second = metrics.ConsumerMetrics("proc-2", "flow-2", "name-2") + first = metrics.ConsumerMetrics("proc", "cons", workspace="ws", flow="fl") + second = metrics.ConsumerMetrics("proc-2", "cons-2") assert enum_factory.call_count == 1 assert histogram_factory.call_count == 1 @@ -96,7 +96,7 @@ def test_producer_metrics_increments_counter_once(monkeypatch): counter_factory.return_value.labels.return_value = labels monkeypatch.setattr(metrics, "Counter", counter_factory) - producer_metrics = metrics.ProducerMetrics("proc", "flow", "output") + producer_metrics = metrics.ProducerMetrics("proc", "output") producer_metrics.inc() counter_factory.assert_called_once() @@ -133,7 +133,7 @@ def test_subscriber_metrics_tracks_received_state_and_dropped(monkeypatch): monkeypatch.setattr(metrics, "Enum", enum_factory) monkeypatch.setattr(metrics, "Counter", counter_factory) - subscriber_metrics = metrics.SubscriberMetrics("proc", "flow", "input") + subscriber_metrics = metrics.SubscriberMetrics("proc", "input") subscriber_metrics.received() subscriber_metrics.state("running") subscriber_metrics.dropped("ignored") diff --git a/trustgraph-base/trustgraph/base/agent_service.py b/trustgraph-base/trustgraph/base/agent_service.py index 06190546..f5828426 100644 --- a/trustgraph-base/trustgraph/base/agent_service.py +++ b/trustgraph-base/trustgraph/base/agent_service.py @@ -8,7 +8,6 @@ from argparse import ArgumentParser import time import logging -from prometheus_client import Histogram from .. schema import AgentRequest, AgentResponse, Error from .. exceptions import TooManyRequests diff --git a/trustgraph-base/trustgraph/base/async_processor.py b/trustgraph-base/trustgraph/base/async_processor.py index b65074bb..c71102bd 100644 --- a/trustgraph-base/trustgraph/base/async_processor.py +++ b/trustgraph-base/trustgraph/base/async_processor.py @@ -84,7 +84,7 @@ class AsyncProcessor: config_subscriber_id = str(uuid.uuid4()) config_consumer_metrics = ConsumerMetrics( - processor = self.id, flow = None, name = "config", + processor=self.id, consumer="config", ) # Subscribe to config notify queue @@ -112,10 +112,10 @@ class AsyncProcessor: config_rr_id = str(uuid.uuid4()) config_req_metrics = ProducerMetrics( - processor = self.id, flow = None, name = "config-request", + processor=self.id, producer="config-request", ) config_resp_metrics = SubscriberMetrics( - processor = self.id, flow = None, name = "config-response", + processor=self.id, subscriber="config-response", ) return RequestResponse( diff --git a/trustgraph-base/trustgraph/base/audit_publisher.py b/trustgraph-base/trustgraph/base/audit_publisher.py index 83d41357..debd19a2 100644 --- a/trustgraph-base/trustgraph/base/audit_publisher.py +++ b/trustgraph-base/trustgraph/base/audit_publisher.py @@ -21,8 +21,7 @@ class AuditPublisher: schema=AuditEvent, metrics=ProducerMetrics( processor=processor_id or component_name, - flow=None, - name="audit-events", + producer="audit-events", ), ) diff --git a/trustgraph-base/trustgraph/base/consumer_spec.py b/trustgraph-base/trustgraph/base/consumer_spec.py index af072cca..1eaf07f2 100644 --- a/trustgraph-base/trustgraph/base/consumer_spec.py +++ b/trustgraph-base/trustgraph/base/consumer_spec.py @@ -16,7 +16,8 @@ class ConsumerSpec(Spec): def add(self, flow: Any, processor: Any, definition: dict[str, Any]) -> None: consumer_metrics = ConsumerMetrics( - processor = flow.id, flow = flow.name, name = self.name, + processor=flow.id, consumer=self.name, + workspace=flow.workspace, flow=flow.name, ) consumer = Consumer( diff --git a/trustgraph-base/trustgraph/base/dynamic_tool_service.py b/trustgraph-base/trustgraph/base/dynamic_tool_service.py index 00a457de..a0633112 100644 --- a/trustgraph-base/trustgraph/base/dynamic_tool_service.py +++ b/trustgraph-base/trustgraph/base/dynamic_tool_service.py @@ -63,7 +63,7 @@ class DynamicToolService(AsyncProcessor): # Create consumer for requests consumer_metrics = ConsumerMetrics( - processor=self.id, flow=None, name="request" + processor=self.id, consumer="request", ) self.consumer = Consumer( @@ -79,7 +79,7 @@ class DynamicToolService(AsyncProcessor): # Create producer for responses producer_metrics = ProducerMetrics( - processor=self.id, flow=None, name="response" + processor=self.id, producer="response", ) self.producer = Producer( @@ -93,7 +93,7 @@ class DynamicToolService(AsyncProcessor): __class__.tool_service_metric = Counter( 'dynamic_tool_service_invocation_count', 'Dynamic tool service invocation count', - ["id"], + ["processor"], ) async def start(self): @@ -133,7 +133,7 @@ class DynamicToolService(AsyncProcessor): ) __class__.tool_service_metric.labels( - id=self.id, + processor=self.id, ).inc() except TooManyRequests as e: diff --git a/trustgraph-base/trustgraph/base/embeddings_service.py b/trustgraph-base/trustgraph/base/embeddings_service.py index 0f5e6237..02c0ce7c 100644 --- a/trustgraph-base/trustgraph/base/embeddings_service.py +++ b/trustgraph-base/trustgraph/base/embeddings_service.py @@ -8,7 +8,6 @@ from argparse import ArgumentParser import time import logging -from prometheus_client import Histogram from .. schema import EmbeddingsRequest, EmbeddingsResponse, Error from .. exceptions import TooManyRequests diff --git a/trustgraph-base/trustgraph/base/image_to_text_service.py b/trustgraph-base/trustgraph/base/image_to_text_service.py index 604521e0..bf504c00 100644 --- a/trustgraph-base/trustgraph/base/image_to_text_service.py +++ b/trustgraph-base/trustgraph/base/image_to_text_service.py @@ -75,7 +75,7 @@ class ImageToTextService(FlowProcessor): __class__.image_to_text_metric = Histogram( 'image_to_text_duration', 'Image-to-text duration (seconds)', - ["id", "flow"], + ["processor"], buckets=[ 0.25, 0.5, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0, 11.0, 12.0, 13.0, 14.0, 15.0, 16.0, @@ -89,7 +89,7 @@ class ImageToTextService(FlowProcessor): __class__.image_to_text_model_metric = Info( 'image_to_text_model', 'Image-to-text model', - ["processor", "flow"] + ["processor"] ) async def on_request(self, msg, consumer, flow): @@ -105,8 +105,7 @@ class ImageToTextService(FlowProcessor): model = flow("model") with __class__.image_to_text_metric.labels( - id=self.id, - flow=f"{flow.name}-{consumer.name}", + processor=self.id, ).time(): response = await self.describe_image( @@ -126,8 +125,7 @@ class ImageToTextService(FlowProcessor): ) __class__.image_to_text_model_metric.labels( - processor = self.id, - flow = flow.name + processor=self.id, ).info({ "model": str(model) if model is not None else "", }) diff --git a/trustgraph-base/trustgraph/base/librarian_client.py b/trustgraph-base/trustgraph/base/librarian_client.py index 2345d2e3..b82af445 100644 --- a/trustgraph-base/trustgraph/base/librarian_client.py +++ b/trustgraph-base/trustgraph/base/librarian_client.py @@ -43,10 +43,12 @@ class LibrarianClient: "librarian_subscriber", f"{id}-librarian", ) + workspace = params.get("workspace") flow_name = params.get("flow_name") librarian_request_metrics = ProducerMetrics( - processor=id, flow=flow_name, name="librarian-request", + processor=id, producer="librarian-request", + workspace=workspace, flow=flow_name, ) self._producer = Producer( @@ -57,7 +59,8 @@ class LibrarianClient: ) librarian_response_metrics = ConsumerMetrics( - processor=id, flow=flow_name, name="librarian-response", + processor=id, consumer="librarian-response", + workspace=workspace, flow=flow_name, ) self._consumer = Consumer( diff --git a/trustgraph-base/trustgraph/base/librarian_spec.py b/trustgraph-base/trustgraph/base/librarian_spec.py index 9f2582ac..af0a8242 100644 --- a/trustgraph-base/trustgraph/base/librarian_spec.py +++ b/trustgraph-base/trustgraph/base/librarian_spec.py @@ -25,6 +25,7 @@ class LibrarianSpec(Spec): processor.id + "--" + flow.workspace + "--" + flow.name + "--librarian--" + str(uuid.uuid4()) ), + workspace=flow.workspace, flow_name=flow.name, ) diff --git a/trustgraph-base/trustgraph/base/llm_service.py b/trustgraph-base/trustgraph/base/llm_service.py index cad1875d..cf5730db 100644 --- a/trustgraph-base/trustgraph/base/llm_service.py +++ b/trustgraph-base/trustgraph/base/llm_service.py @@ -94,7 +94,7 @@ class LlmService(FlowProcessor): __class__.text_completion_metric = Histogram( 'text_completion_duration', 'Text completion duration (seconds)', - ["id", "flow"], + ["processor"], buckets=[ 0.25, 0.5, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0, 11.0, 12.0, 13.0, 14.0, 15.0, 16.0, @@ -108,7 +108,7 @@ class LlmService(FlowProcessor): __class__.text_completion_model_metric = Info( 'text_completion_model', 'Text completion model', - ["processor", "flow"] + ["processor"] ) async def on_request(self, msg, consumer, flow): @@ -133,8 +133,7 @@ class LlmService(FlowProcessor): # Streaming mode with __class__.text_completion_metric.labels( - id=self.id, - flow=f"{flow.name}-{consumer.name}", + processor=self.id, ).time(): async for chunk in self.generate_content_stream( @@ -157,8 +156,7 @@ class LlmService(FlowProcessor): # Non-streaming mode (original behavior) with __class__.text_completion_metric.labels( - id=self.id, - flow=f"{flow.name}-{consumer.name}", + processor=self.id, ).time(): response = await self.generate_content( @@ -179,8 +177,7 @@ class LlmService(FlowProcessor): ) __class__.text_completion_model_metric.labels( - processor = self.id, - flow = flow.name + processor=self.id, ).info({ "model": str(model) if model is not None else "", "temperature": str(temperature) if temperature is not None else "", diff --git a/trustgraph-base/trustgraph/base/metrics.py b/trustgraph-base/trustgraph/base/metrics.py index f9cf3441..f3f5affc 100644 --- a/trustgraph-base/trustgraph/base/metrics.py +++ b/trustgraph-base/trustgraph/base/metrics.py @@ -6,82 +6,85 @@ from prometheus_client import start_http_server, Info, Enum, Histogram from prometheus_client import Counter class ConsumerMetrics: - """ - Metrics tracking and reporting for flow consumers. - - This class manages prometheus metrics specifically related to consumers - within the flow, including state, requests, processing time, and queues. - """ - def __init__(self, processor: str, flow: str, name: str) -> None: + def __init__(self, processor: str, consumer: str, + workspace: str | None = None, + flow: str | None = None) -> None: self.processor = processor + self.consumer = consumer + self.workspace = workspace self.flow = flow - self.name = name if not hasattr(__class__, "state_metric"): __class__.state_metric = Enum( 'consumer_state', 'Consumer state', - ["processor", "flow", "name"], + ["processor", "workspace", "flow", "consumer"], states=['stopped', 'running'] ) if not hasattr(__class__, "request_metric"): __class__.request_metric = Histogram( 'request_latency', 'Request latency (seconds)', - ["processor", "flow", "name"], + ["processor", "consumer"], ) if not hasattr(__class__, "processing_metric"): __class__.processing_metric = Counter( 'processing_count', 'Processing count', - ["processor", "flow", "name", "status"], + ["processor", "workspace", "flow", "consumer", "status"], ) if not hasattr(__class__, "rate_limit_metric"): __class__.rate_limit_metric = Counter( 'rate_limit_count', 'Rate limit event count', - ["processor", "flow", "name"], + ["processor", "workspace", "flow", "consumer"], ) def process(self, status: str) -> None: __class__.processing_metric.labels( - processor = self.processor, flow = self.flow, name = self.name, - status=status + processor=self.processor, workspace=self.workspace, + flow=self.flow, consumer=self.consumer, status=status, ).inc() def rate_limit(self) -> None: __class__.rate_limit_metric.labels( - processor = self.processor, flow = self.flow, name = self.name, + processor=self.processor, workspace=self.workspace, + flow=self.flow, consumer=self.consumer, ).inc() def state(self, state: str) -> None: __class__.state_metric.labels( - processor = self.processor, flow = self.flow, name = self.name, + processor=self.processor, workspace=self.workspace, + flow=self.flow, consumer=self.consumer, ).state(state) def record_time(self) -> Any: return __class__.request_metric.labels( - processor = self.processor, flow = self.flow, name = self.name, + processor=self.processor, consumer=self.consumer, ).time() class ProducerMetrics: - def __init__(self, processor: str, flow: str, name: str) -> None: + def __init__(self, processor: str, producer: str, + workspace: str | None = None, + flow: str | None = None) -> None: self.processor = processor + self.producer = producer + self.workspace = workspace self.flow = flow - self.name = name if not hasattr(__class__, "producer_metric"): __class__.producer_metric = Counter( 'producer_count', 'Output items produced', - ["processor", "flow", "name"], + ["processor", "workspace", "flow", "producer"], ) def inc(self) -> None: __class__.producer_metric.labels( - processor = self.processor, flow = self.flow, name = self.name + processor=self.processor, workspace=self.workspace, + flow=self.flow, producer=self.producer, ).inc() class ProcessorMetrics: @@ -102,43 +105,48 @@ class ProcessorMetrics: class SubscriberMetrics: - def __init__(self, processor: str, flow: str, name: str) -> None: + def __init__(self, processor: str, subscriber: str, + workspace: str | None = None, + flow: str | None = None) -> None: self.processor = processor + self.subscriber = subscriber + self.workspace = workspace self.flow = flow - self.name = name if not hasattr(__class__, "state_metric"): __class__.state_metric = Enum( 'subscriber_state', 'Subscriber state', - ["processor", "flow", "name"], + ["processor", "workspace", "flow", "subscriber"], states=['stopped', 'running'] ) if not hasattr(__class__, "received_metric"): __class__.received_metric = Counter( 'received_count', 'Received count', - ["processor", "flow", "name"], + ["processor", "workspace", "flow", "subscriber"], ) if not hasattr(__class__, "dropped_metric"): __class__.dropped_metric = Counter( 'dropped_count', 'Dropped messages count', - ["processor", "flow", "name"], + ["processor", "workspace", "flow", "subscriber"], ) def received(self) -> None: __class__.received_metric.labels( - processor = self.processor, flow = self.flow, name = self.name, + processor=self.processor, workspace=self.workspace, + flow=self.flow, subscriber=self.subscriber, ).inc() def state(self, state: str) -> None: - __class__.state_metric.labels( - processor = self.processor, flow = self.flow, name = self.name, + processor=self.processor, workspace=self.workspace, + flow=self.flow, subscriber=self.subscriber, ).state(state) def dropped(self, state: str) -> None: __class__.dropped_metric.labels( - processor = self.processor, flow = self.flow, name = self.name, + processor=self.processor, workspace=self.workspace, + flow=self.flow, subscriber=self.subscriber, ).inc() diff --git a/trustgraph-base/trustgraph/base/producer_spec.py b/trustgraph-base/trustgraph/base/producer_spec.py index 16905f4b..55b3f6f3 100644 --- a/trustgraph-base/trustgraph/base/producer_spec.py +++ b/trustgraph-base/trustgraph/base/producer_spec.py @@ -14,7 +14,8 @@ class ProducerSpec(Spec): def add(self, flow: Any, processor: Any, definition: dict[str, Any]) -> None: producer_metrics = ProducerMetrics( - processor = flow.id, flow = flow.name, name = self.name + processor=flow.id, producer=self.name, + workspace=flow.workspace, flow=flow.name, ) producer = Producer( diff --git a/trustgraph-base/trustgraph/base/request_response_spec.py b/trustgraph-base/trustgraph/base/request_response_spec.py index 7c8ad3bd..1dd060e8 100644 --- a/trustgraph-base/trustgraph/base/request_response_spec.py +++ b/trustgraph-base/trustgraph/base/request_response_spec.py @@ -132,11 +132,13 @@ class RequestResponseSpec(Spec): return request_metrics = ProducerMetrics( - processor = flow.id, flow = flow.name, name = self.request_name + processor=flow.id, producer=self.request_name, + workspace=flow.workspace, flow=flow.name, ) response_metrics = SubscriberMetrics( - processor = flow.id, flow = flow.name, name = self.request_name + processor=flow.id, subscriber=self.response_name, + workspace=flow.workspace, flow=flow.name, ) rr = self.impl( diff --git a/trustgraph-base/trustgraph/base/subscriber_spec.py b/trustgraph-base/trustgraph/base/subscriber_spec.py index 80f9b0d5..ba965a2f 100644 --- a/trustgraph-base/trustgraph/base/subscriber_spec.py +++ b/trustgraph-base/trustgraph/base/subscriber_spec.py @@ -15,7 +15,8 @@ class SubscriberSpec(Spec): def add(self, flow: Any, processor: Any, definition: dict[str, Any]) -> None: subscriber_metrics = SubscriberMetrics( - processor = flow.id, flow = flow.name, name = self.name + processor=flow.id, subscriber=self.name, + workspace=flow.workspace, flow=flow.name, ) subscriber = Subscriber( diff --git a/trustgraph-base/trustgraph/base/tool_service.py b/trustgraph-base/trustgraph/base/tool_service.py index eeaced6a..2506cff1 100644 --- a/trustgraph-base/trustgraph/base/tool_service.py +++ b/trustgraph-base/trustgraph/base/tool_service.py @@ -50,7 +50,7 @@ class ToolService(FlowProcessor): if not hasattr(__class__, "tool_invocation_metric"): __class__.tool_invocation_metric = Counter( 'tool_invocation_count', 'Tool invocation count', - ["id", "flow", "name"], + ["processor", "workspace", "flow", "tool"], ) async def on_request(self, msg, consumer, flow): @@ -89,7 +89,8 @@ class ToolService(FlowProcessor): ) __class__.tool_invocation_metric.labels( - id = self.id, flow = flow.name, name = request.name, + processor=self.id, workspace=flow.workspace, + flow=flow.name, tool=request.name, ).inc() except TooManyRequests as e: diff --git a/trustgraph-flow/trustgraph/agent/react/tools.py b/trustgraph-flow/trustgraph/agent/react/tools.py index ae9507ab..c00e36ff 100644 --- a/trustgraph-flow/trustgraph/agent/react/tools.py +++ b/trustgraph-flow/trustgraph/agent/react/tools.py @@ -277,13 +277,11 @@ class ToolServiceImpl: request_metrics = ProducerMetrics( processor=self.processor.id, - flow="tool-service", - name=self.request_queue + producer=self.request_queue, ) response_metrics = SubscriberMetrics( processor=self.processor.id, - flow="tool-service", - name=self.response_queue + subscriber=self.response_queue, ) # Create unique subscription for responses diff --git a/trustgraph-flow/trustgraph/bootstrap/bootstrapper/service.py b/trustgraph-flow/trustgraph/bootstrap/bootstrapper/service.py index 81b7e98d..f97126ca 100644 --- a/trustgraph-flow/trustgraph/bootstrap/bootstrapper/service.py +++ b/trustgraph-flow/trustgraph/bootstrap/bootstrapper/service.py @@ -173,12 +173,12 @@ class Processor(AsyncProcessor): request_topic=config_request_queue, request_schema=ConfigRequest, request_metrics=ProducerMetrics( - processor=self.id, flow=None, name="config-request", + processor=self.id, producer="config-request", ), response_topic=config_response_queue, response_schema=ConfigResponse, response_metrics=SubscriberMetrics( - processor=self.id, flow=None, name="config-response", + processor=self.id, subscriber="config-response", ), ) @@ -191,12 +191,12 @@ class Processor(AsyncProcessor): request_topic=f"{flow_request_queue}:{workspace}", request_schema=FlowRequest, request_metrics=ProducerMetrics( - processor=self.id, flow=None, name="flow-request", + processor=self.id, producer="flow-request", ), response_topic=f"{flow_response_queue}:{workspace}", response_schema=FlowResponse, response_metrics=SubscriberMetrics( - processor=self.id, flow=None, name="flow-response", + processor=self.id, subscriber="flow-response", ), ) @@ -209,12 +209,12 @@ class Processor(AsyncProcessor): request_topic=iam_request_queue, request_schema=IamRequest, request_metrics=ProducerMetrics( - processor=self.id, flow=None, name="iam-request", + processor=self.id, producer="iam-request", ), response_topic=iam_response_queue, response_schema=IamResponse, response_metrics=SubscriberMetrics( - processor=self.id, flow=None, name="iam-response", + processor=self.id, subscriber="iam-response", ), ) diff --git a/trustgraph-flow/trustgraph/chunking/recursive/chunker.py b/trustgraph-flow/trustgraph/chunking/recursive/chunker.py index 33f83e5a..b6d64994 100755 --- a/trustgraph-flow/trustgraph/chunking/recursive/chunker.py +++ b/trustgraph-flow/trustgraph/chunking/recursive/chunker.py @@ -53,7 +53,7 @@ class Processor(ChunkingService): if not hasattr(__class__, "chunk_metric"): __class__.chunk_metric = Histogram( 'chunk_size', 'Chunk size', - ["id", "flow"], + ["processor"], buckets=[100, 160, 250, 400, 650, 1000, 1600, 2500, 4000, 6400, 10000, 16000] ) @@ -184,7 +184,7 @@ class Processor(ChunkingService): ) __class__.chunk_metric.labels( - id=consumer.id, flow=consumer.flow + processor=self.id, ).observe(chunk_length) await flow("output").send(r) diff --git a/trustgraph-flow/trustgraph/chunking/token/chunker.py b/trustgraph-flow/trustgraph/chunking/token/chunker.py index d099d824..c28f773c 100755 --- a/trustgraph-flow/trustgraph/chunking/token/chunker.py +++ b/trustgraph-flow/trustgraph/chunking/token/chunker.py @@ -51,7 +51,7 @@ class Processor(ChunkingService): if not hasattr(__class__, "chunk_metric"): __class__.chunk_metric = Histogram( 'chunk_size', 'Chunk size', - ["id", "flow"], + ["processor"], buckets=[100, 160, 250, 400, 650, 1000, 1600, 2500, 4000, 6400, 10000, 16000] ) @@ -180,7 +180,7 @@ class Processor(ChunkingService): ) __class__.chunk_metric.labels( - id=consumer.id, flow=consumer.flow + processor=self.id, ).observe(chunk_length) await flow("output").send(r) diff --git a/trustgraph-flow/trustgraph/config/service/service.py b/trustgraph-flow/trustgraph/config/service/service.py index 725f1106..eaaaee65 100644 --- a/trustgraph-flow/trustgraph/config/service/service.py +++ b/trustgraph-flow/trustgraph/config/service/service.py @@ -106,13 +106,13 @@ class Processor(AsyncProcessor): ) config_request_metrics = ConsumerMetrics( - processor = self.id, flow = None, name = "config-request" + processor=self.id, consumer="config-request", ) config_response_metrics = ProducerMetrics( - processor = self.id, flow = None, name = "config-response" + processor=self.id, producer="config-response", ) config_push_metrics = ProducerMetrics( - processor = self.id, flow = None, name = "config-push" + processor=self.id, producer="config-push", ) self.config_request_queue_base = config_request_queue @@ -225,8 +225,8 @@ class Processor(AsyncProcessor): topic=resp_queue, schema=ConfigResponse, metrics=ProducerMetrics( - processor=self.id, flow=None, - name=f"config-response-{workspace_id}", + processor=self.id, producer="config-response", + workspace=workspace_id, ), ) @@ -242,8 +242,8 @@ class Processor(AsyncProcessor): workspace=workspace_id, ), metrics=ConsumerMetrics( - processor=self.id, flow=None, - name=f"config-request-{workspace_id}", + processor=self.id, consumer="config-request", + workspace=workspace_id, ), ) diff --git a/trustgraph-flow/trustgraph/cores/service.py b/trustgraph-flow/trustgraph/cores/service.py index 9d656c6d..fb8ee08e 100755 --- a/trustgraph-flow/trustgraph/cores/service.py +++ b/trustgraph-flow/trustgraph/cores/service.py @@ -120,8 +120,8 @@ class Processor(WorkspaceProcessor): topic=resp_queue, schema=KnowledgeResponse, metrics=ProducerMetrics( - processor=self.id, flow=None, - name=f"knowledge-response-{workspace}", + processor=self.id, producer="knowledge-response", + workspace=workspace, ), ) @@ -136,8 +136,8 @@ class Processor(WorkspaceProcessor): self.on_knowledge_request, workspace=workspace, ), metrics=ConsumerMetrics( - processor=self.id, flow=None, - name=f"knowledge-request-{workspace}", + processor=self.id, consumer="knowledge-request", + workspace=workspace, ), ) diff --git a/trustgraph-flow/trustgraph/flow/service/service.py b/trustgraph-flow/trustgraph/flow/service/service.py index 36362b7e..6ddd1b5e 100644 --- a/trustgraph-flow/trustgraph/flow/service/service.py +++ b/trustgraph-flow/trustgraph/flow/service/service.py @@ -55,10 +55,10 @@ class Processor(WorkspaceProcessor): ) config_req_metrics = ProducerMetrics( - processor=self.id, flow=None, name="config-request", + processor=self.id, producer="config-request", ) config_resp_metrics = SubscriberMetrics( - processor=self.id, flow=None, name="config-response", + processor=self.id, subscriber="config-response", ) config_rr_id = str(uuid.uuid4()) @@ -100,8 +100,8 @@ class Processor(WorkspaceProcessor): topic=resp_queue, schema=FlowResponse, metrics=ProducerMetrics( - processor=self.id, flow=None, - name=f"flow-response-{workspace}", + processor=self.id, producer="flow-response", + workspace=workspace, ), ) @@ -116,8 +116,8 @@ class Processor(WorkspaceProcessor): self.on_flow_request, workspace=workspace, ), metrics=ConsumerMetrics( - processor=self.id, flow=None, - name=f"flow-request-{workspace}", + processor=self.id, consumer="flow-request", + workspace=workspace, ), ) diff --git a/trustgraph-flow/trustgraph/gateway/auth.py b/trustgraph-flow/trustgraph/gateway/auth.py index f0e06911..a343fd26 100644 --- a/trustgraph-flow/trustgraph/gateway/auth.py +++ b/trustgraph-flow/trustgraph/gateway/auth.py @@ -164,12 +164,12 @@ class IamAuth: request_topic=iam_request_queue, request_schema=IamRequest, request_metrics=ProducerMetrics( - processor=self.id, flow=None, name="iam-request", + processor=self.id, producer="iam-request", ), response_topic=iam_response_queue, response_schema=IamResponse, response_metrics=SubscriberMetrics( - processor=self.id, flow=None, name="iam-response", + processor=self.id, subscriber="iam-response", ), ) diff --git a/trustgraph-flow/trustgraph/gateway/config/receiver.py b/trustgraph-flow/trustgraph/gateway/config/receiver.py index 8c42381f..e70670c6 100755 --- a/trustgraph-flow/trustgraph/gateway/config/receiver.py +++ b/trustgraph-flow/trustgraph/gateway/config/receiver.py @@ -96,12 +96,10 @@ class ConfigReceiver: id = str(uuid.uuid4()) config_req_metrics = ProducerMetrics( - processor="api-gateway", flow=None, - name="config-request", + processor="api-gateway", producer="config-request", ) config_resp_metrics = SubscriberMetrics( - processor="api-gateway", flow=None, - name="config-response", + processor="api-gateway", subscriber="config-response", ) return RequestResponse( diff --git a/trustgraph-flow/trustgraph/iam/noauth/service.py b/trustgraph-flow/trustgraph/iam/noauth/service.py index 76d13a3c..833ed720 100644 --- a/trustgraph-flow/trustgraph/iam/noauth/service.py +++ b/trustgraph-flow/trustgraph/iam/noauth/service.py @@ -44,10 +44,10 @@ class Processor(AsyncProcessor): super().__init__(**params) iam_request_metrics = ConsumerMetrics( - processor=self.id, flow=None, name="iam-request", + processor=self.id, consumer="iam-request", ) iam_response_metrics = ProducerMetrics( - processor=self.id, flow=None, name="iam-response", + processor=self.id, producer="iam-response", ) self.iam_request_topic = iam_req_q @@ -88,10 +88,10 @@ class Processor(AsyncProcessor): def _create_config_client(self): config_rr_id = str(uuid.uuid4()) config_req_metrics = ProducerMetrics( - processor=self.id, flow=None, name="config-request", + processor=self.id, producer="config-request", ) config_resp_metrics = SubscriberMetrics( - processor=self.id, flow=None, name="config-response", + processor=self.id, subscriber="config-response", ) return RequestResponse( backend=self.pubsub, diff --git a/trustgraph-flow/trustgraph/iam/service/service.py b/trustgraph-flow/trustgraph/iam/service/service.py index 52498230..215215f3 100644 --- a/trustgraph-flow/trustgraph/iam/service/service.py +++ b/trustgraph-flow/trustgraph/iam/service/service.py @@ -120,10 +120,10 @@ class Processor(AsyncProcessor): ) iam_request_metrics = ConsumerMetrics( - processor=self.id, flow=None, name="iam-request", + processor=self.id, consumer="iam-request", ) iam_response_metrics = ProducerMetrics( - processor=self.id, flow=None, name="iam-response", + processor=self.id, producer="iam-response", ) self.iam_request_topic = iam_req_q @@ -179,10 +179,10 @@ class Processor(AsyncProcessor): import uuid config_rr_id = str(uuid.uuid4()) config_req_metrics = ProducerMetrics( - processor=self.id, flow=None, name="config-request", + processor=self.id, producer="config-request", ) config_resp_metrics = SubscriberMetrics( - processor=self.id, flow=None, name="config-response", + processor=self.id, subscriber="config-response", ) return RequestResponse( backend=self.pubsub, diff --git a/trustgraph-flow/trustgraph/librarian/service.py b/trustgraph-flow/trustgraph/librarian/service.py index 4d3efbfb..c2347e3c 100755 --- a/trustgraph-flow/trustgraph/librarian/service.py +++ b/trustgraph-flow/trustgraph/librarian/service.py @@ -171,7 +171,7 @@ class Processor(WorkspaceProcessor): # Config service client for collection management config_request_metrics = ProducerMetrics( - processor = id, flow = None, name = "config-request" + processor=id, producer="config-request", ) self.config_request_producer = Producer( @@ -182,7 +182,7 @@ class Processor(WorkspaceProcessor): ) config_response_metrics = ConsumerMetrics( - processor = id, flow = None, name = "config-response" + processor=id, consumer="config-response", ) self.config_response_consumer = Consumer( @@ -258,8 +258,8 @@ class Processor(WorkspaceProcessor): topic=lib_resp_queue, schema=LibrarianResponse, metrics=ProducerMetrics( - processor=self.id, flow=None, - name=f"librarian-response-{workspace}", + processor=self.id, producer="librarian-response", + workspace=workspace, ), ) @@ -268,8 +268,8 @@ class Processor(WorkspaceProcessor): topic=col_resp_queue, schema=CollectionManagementResponse, metrics=ProducerMetrics( - processor=self.id, flow=None, - name=f"collection-response-{workspace}", + processor=self.id, producer="collection-response", + workspace=workspace, ), ) @@ -284,8 +284,8 @@ class Processor(WorkspaceProcessor): self.on_librarian_request, workspace=workspace, ), metrics=ConsumerMetrics( - processor=self.id, flow=None, - name=f"librarian-request-{workspace}", + processor=self.id, consumer="librarian-request", + workspace=workspace, ), ) @@ -300,8 +300,8 @@ class Processor(WorkspaceProcessor): self.on_collection_request, workspace=workspace, ), metrics=ConsumerMetrics( - processor=self.id, flow=None, - name=f"collection-request-{workspace}", + processor=self.id, consumer="collection-request", + workspace=workspace, ), ) diff --git a/trustgraph-flow/trustgraph/model/text_completion/azure/llm.py b/trustgraph-flow/trustgraph/model/text_completion/azure/llm.py index 00056539..4476c623 100755 --- a/trustgraph-flow/trustgraph/model/text_completion/azure/llm.py +++ b/trustgraph-flow/trustgraph/model/text_completion/azure/llm.py @@ -6,7 +6,6 @@ serverless endpoint service. Input is prompt, output is response. import requests import json -from prometheus_client import Histogram import os import logging diff --git a/trustgraph-flow/trustgraph/model/text_completion/azure_openai/llm.py b/trustgraph-flow/trustgraph/model/text_completion/azure_openai/llm.py index f77fb281..d0f50123 100755 --- a/trustgraph-flow/trustgraph/model/text_completion/azure_openai/llm.py +++ b/trustgraph-flow/trustgraph/model/text_completion/azure_openai/llm.py @@ -5,7 +5,6 @@ OpenAI endpoit service. Input is prompt, output is response. """ import json -from prometheus_client import Histogram from openai import AzureOpenAI, RateLimitError import os import logging diff --git a/trustgraph-flow/trustgraph/model/text_completion/cohere/llm.py b/trustgraph-flow/trustgraph/model/text_completion/cohere/llm.py index 10c72d85..f0395c0d 100755 --- a/trustgraph-flow/trustgraph/model/text_completion/cohere/llm.py +++ b/trustgraph-flow/trustgraph/model/text_completion/cohere/llm.py @@ -6,7 +6,6 @@ Input is prompt, output is response. import cohere from cohere.errors import TooManyRequestsError, ServiceUnavailableError -from prometheus_client import Histogram import os import logging