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

Add a Prom metric to consumers & consumer/producers to track the running

Browse source 
state.
This commit is contained in:
Cyber MacGeddon 2024-09-29 23:45:33 +01:00
parent efc364583b
commit 39579e1b84
2 changed files with 24 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

13
trustgraph/base/consumer.py
View file

@ -1,6 +1,6 @@
from pulsar.schema import JsonSchema
from prometheus_client import start_http_server, Histogram, Info, Counter
from prometheus_client import Histogram, Info, Counter, Enum
import time
from . base_processor import BaseProcessor
@ -10,6 +10,15 @@ class Consumer(BaseProcessor):
def __init__(self, **params):
if not hasattr(__class__, "state_metric"):
__class__.state_metric = Enum(
'processor_state', 'Processor state',
states=['starting', 'running', 'stopped']
)
__class__.state_metric.state('starting')
__class__.state_metric.state('starting')
super(Consumer, self).__init__(**params)
input_queue = params.get("input_queue")
@ -47,6 +56,8 @@ class Consumer(BaseProcessor):
def run(self):
__class__.state_metric.state('running')
while True:
msg = self.consumer.receive()

13
trustgraph/base/consumer_producer.py
View file

@ -1,6 +1,6 @@
from pulsar.schema import JsonSchema
from prometheus_client import Histogram, Info, Counter
from prometheus_client import Histogram, Info, Counter, Enum
import time
from . base_processor import BaseProcessor
@ -12,6 +12,15 @@ class ConsumerProducer(BaseProcessor):
def __init__(self, **params):
if not hasattr(__class__, "state_metric"):
__class__.state_metric = Enum(
'processor_state', 'Processor state',
states=['starting', 'running', 'stopped']
)
__class__.state_metric.state('starting')
__class__.state_metric.state('starting')
input_queue = params.get("input_queue")
output_queue = params.get("output_queue")
subscriber = params.get("subscriber")
@ -66,6 +75,8 @@ class ConsumerProducer(BaseProcessor):
def run(self):
__class__.state_metric.state('running')
while True:
msg = self.consumer.receive()