apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-20 18:51:03 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Make metrics uniform

Browse source
This commit is contained in:
Cyber MacGeddon 2025-04-23 10:46:38 +01:00
parent 0fbe6f6676
commit 3c5f6edc54
7 changed files with 48 additions and 28 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -33,7 +33,7 @@ class AsyncProcessor:
self.pulsar_client = PulsarClient(**params)
# Initialise metrics, records the parameters
ProcessorMetrics(id=self.id).info({
ProcessorMetrics(processor = self.id).info({
k: str(params[k])
for k in params
if k != "id"

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

@ -12,7 +12,7 @@ class ConsumerSpec(Spec):
def add(self, flow, processor, definition):
consumer_metrics = ConsumerMetrics(
flow.id, f"{flow.name}-{self.name}"
processor = flow.id, flow = flow.name, name = self.name,
)
consumer = Consumer(

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

@ -4,79 +4,89 @@ from prometheus_client import Counter
class ConsumerMetrics:
def __init__(self, id, flow=None):
def __init__(self, processor, flow, name):
self.id = id
self.processor = processor
self.flow = flow
self.name = name
if not hasattr(__class__, "state_metric"):
__class__.state_metric = Enum(
'consumer_state', 'Consumer state',
["id", "flow"],
["processor", "flow", "name"],
states=['stopped', 'running']
)
if not hasattr(__class__, "request_metric"):
__class__.request_metric = Histogram(
'request_latency', 'Request latency (seconds)',
["id", "flow"],
["processor", "flow", "name"],
)
if not hasattr(__class__, "processing_metric"):
__class__.processing_metric = Counter(
'processing_count', 'Processing count',
["id", "flow", "status"]
["processor", "flow", "name"],
)
if not hasattr(__class__, "rate_limit_metric"):
__class__.rate_limit_metric = Counter(
'rate_limit_count', 'Rate limit event count',
["id", "flow"]
["processor", "flow", "name"],
)
def process(self, status):
__class__.processing_metric.labels(
id=self.id, flow=self.flow, status=status
processor = self.processor, flow=self.flow, status=status
).inc()
def rate_limit(self):
__class__.rate_limit_metric.labels(
id=self.id, flow=self.flow
processor = self.processor, flow=self.flow
).inc()
def state(self, state):
__class__.state_metric.labels(
id=self.id, flow=self.flow
processor = self.processor, flow=self.flow
).state(state)
def record_time(self):
return __class__.request_metric.labels(
id=self.id, flow=self.flow
processor = self.processor, flow=self.flow
).time()
class ProducerMetrics:
def __init__(self, id, flow=None):
self.id = id
def __init__(self, processor, flow, name):
self.processor = processor
self.flow = flow
self.name = name
if not hasattr(__class__, "output_metric"):
__class__.output_metric = Counter(
'output_count', 'Output items created',
["id", "flow"]
["processor", "flow", "name"],
)
def inc(self):
__class__.output_metric.labels(id=self.id, flow=self.flow).inc()
__class__.output_metric.labels(
processor = self.processor, flow = self.flow, name = self.name
).inc()
class ProcessorMetrics:
def __init__(self, id):
def __init__(self, processor):
self.id = id
self.processor = processor
if not hasattr(__class__, "processor_metric"):
__class__.processor_metric = Info(
'processor', 'Processor configuration',
["id"]
["processor"]
)
def info(self, info):
__class__.processor_metric.labels(id=self.id).info(info)
__class__.processor_metric.labels(
processor = self.processor
).info(info)

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

@ -11,7 +11,7 @@ class ProducerSpec(Spec):
def add(self, flow, processor, definition):
producer_metrics = ProducerMetrics(
flow.id, f"{flow.name}-{self.name}"
processor = flow.id, flow = flow.name, name = self.name
)
producer = Producer(

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

@ -117,7 +117,7 @@ class RequestResponseSpec(Spec):
def add(self, flow, processor, definition):
producer_metrics = ProducerMetrics(
flow.id, f"{flow.name}-{self.response_name}"
process = flow.id, name = flow.name, name = self.response_name
)
rr = self.impl(

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

@ -13,7 +13,7 @@ class SubscriberSpec(Spec):
# FIXME: Metrics not used
subscriber_metrics = ConsumerMetrics(
flow.id, f"{flow.name}-{self.name}"
processor = flow.id, flow = flow.name, name = self.name
)
subscriber = Subscriber(

20
trustgraph-flow/trustgraph/config/service/service.py
View file

@ -68,12 +68,22 @@ class Processor(AsyncProcessor):
}
)
config_request_metrics = ConsumerMetrics(id + "-config-request")
config_response_metrics = ProducerMetrics(id + "-config-response")
config_push_metrics = ProducerMetrics(id + "-config-push")
config_request_metrics = ConsumerMetrics(
processor = self.id, flow = None, name = "config-request"
)
config_response_metrics = ProducerMetrics(
processor = self.id, flow = None, name = "config-response"
)
config_push_metrics = ProducerMetrics(
processor = self.id, flow = None, name = "config-push"
)
flow_request_metrics = ConsumerMetrics(id + "-flow-request")
flow_response_metrics = ProducerMetrics(id + "-flow-response")
flow_request_metrics = ConsumerMetrics(
processor = self.id, flow = None, name = "flow-request"
)
flow_response_metrics = ProducerMetrics(
processor = self.id, flow = None, name = "flow-response"
)
self.config_request_consumer = Consumer(
taskgroup = self.taskgroup,