apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-27 22:21:04 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

refactor: reduce Prometheus metrics cardinality and clarify label names (#1059)

Browse source 
With many workspaces and flows, the prometheus metrics expand
greatly. Many significant metrics (e.g. latency) make more sense
measured globally. This change introduces a consolidation.

Histograms (request_latency, text_completion_duration, chunk_size, etc.)
are now global with only processor-level labels, eliminating per-workspace
time series multiplication. Counters and enums retain workspace and flow
labels for per-workspace debugging visibility.

Renamed metric labels for clarity: name -> consumer/producer/subscriber,
and added explicit workspace label alongside flow. Fixed bug where
request_response_spec labelled response subscriber with request_name
instead of response_name. Removed unused Histogram imports from model
and service files.
This commit is contained in:
cybermaggedon 2026-07-23 14:51:47 +01:00 • committed by GitHub
parent eb059707f0
commit f7026efeda
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
31 changed files with 129 additions and 126 deletions
Download patch file Download diff file Expand all files Collapse all files

1
trustgraph-base/trustgraph/base/agent_service.py
View file

@ -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

6
trustgraph-base/trustgraph/base/async_processor.py
View file

@ -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(

3
trustgraph-base/trustgraph/base/audit_publisher.py
View file

@ -21,8 +21,7 @@ class AuditPublisher:
schema=AuditEvent,
metrics=ProducerMetrics(
processor=processor_id or component_name,
flow=None,
name="audit-events",
producer="audit-events",
),
)

3
trustgraph-base/trustgraph/base/consumer_spec.py
View file

@ -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(

8
trustgraph-base/trustgraph/base/dynamic_tool_service.py
View file

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

1
trustgraph-base/trustgraph/base/embeddings_service.py
View file

@ -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

10
trustgraph-base/trustgraph/base/image_to_text_service.py
View file

@ -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 "",
})

7
trustgraph-base/trustgraph/base/librarian_client.py
View file

@ -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(

1
trustgraph-base/trustgraph/base/librarian_spec.py
View file

@ -25,6 +25,7 @@ class LibrarianSpec(Spec):
processor.id + "--" + flow.workspace + "--" +
flow.name + "--librarian--" + str(uuid.uuid4())
),
workspace=flow.workspace,
flow_name=flow.name,
)

13
trustgraph-base/trustgraph/base/llm_service.py
View file

@ -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 "",

68
trustgraph-base/trustgraph/base/metrics.py
View file

@ -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()

3
trustgraph-base/trustgraph/base/producer_spec.py
View file

@ -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(

6
trustgraph-base/trustgraph/base/request_response_spec.py
View file

@ -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(

3
trustgraph-base/trustgraph/base/subscriber_spec.py
View file

@ -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(

5
trustgraph-base/trustgraph/base/tool_service.py
View file

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