mirror of
https://github.com/trustgraph-ai/trustgraph.git
synced 2026-07-17 17:21:02 +02:00
Metrics working
This commit is contained in:
Cyber MacGeddon
parent
726baa9932
commit
9fdc408a95
6 changed files with 152 additions and 46 deletions
|
|
@ -5,4 +5,5 @@ from . consumer import Consumer
|
|||
from . producer import Producer
|
||||
from . publisher import Publisher
|
||||
from . subscriber import Subscriber
|
||||
from . metrics import ProcessorMetrics, ConsumerMetrics, ProducerMetrics
|
||||
|
||||
|
|
|
|||
|
|
@ -73,16 +73,25 @@ class AsyncProcessor:
|
|||
while True:
|
||||
await asyncio.sleep(2)
|
||||
|
||||
def subscribe(self, queue, subscriber, schema, handler):
|
||||
def subscribe(self, queue, subscriber, schema, handler, metrics=None):
|
||||
|
||||
return Consumer(
|
||||
self.taskgroup, self.client, queue, subscriber, schema, handler
|
||||
taskgroup = self.taskgroup,
|
||||
client = self.client,
|
||||
queue = queue,
|
||||
subscriber = subscriber,
|
||||
schema = schema,
|
||||
handler = handler,
|
||||
metrics = metrics,
|
||||
)
|
||||
|
||||
def publish(self, queue, schema):
|
||||
def publish(self, queue, schema, metrics=None):
|
||||
|
||||
return Producer(
|
||||
self.client, queue, schema
|
||||
client = self.client,
|
||||
queue = queue,
|
||||
schema = schema,
|
||||
metrics = metrics,
|
||||
)
|
||||
|
||||
def set_processor_state(self, flow, state):
|
||||
|
|
@ -166,7 +175,7 @@ class AsyncProcessor:
|
|||
print("Exception group:")
|
||||
|
||||
for se in e.exceptions:
|
||||
print(" Type:", type(se))
|
||||
print(" Type:", type(se))
|
||||
print(f" Exception: {se}")
|
||||
|
||||
except Exception as e:
|
||||
|
|
|
|||
|
|
@ -8,7 +8,8 @@ class Consumer:
|
|||
|
||||
def __init__(
|
||||
self, taskgroup, client, queue, subscriber, schema,
|
||||
handler
|
||||
handler, rate_limit_retry_time = 10,
|
||||
rate_limit_timeout = 7200, metrics = None
|
||||
):
|
||||
|
||||
self.taskgroup = taskgroup
|
||||
|
|
@ -17,10 +18,14 @@ class Consumer:
|
|||
self.subscriber = subscriber
|
||||
self.schema = schema
|
||||
self.handler = handler
|
||||
self.rate_limit_retry_time = rate_limit_retry_time
|
||||
self.rate_limit_timeout = rate_limit_timeout
|
||||
|
||||
self.running = True
|
||||
self.task = None
|
||||
|
||||
self.metrics = metrics
|
||||
|
||||
async def start(self):
|
||||
|
||||
self.running = True
|
||||
|
|
@ -29,17 +34,23 @@ class Consumer:
|
|||
self.queue, self.subscriber, self.schema
|
||||
)
|
||||
|
||||
# Puts it in the stopped state, the run thread should set running
|
||||
if self.metrics:
|
||||
self.metrics.state("stopped")
|
||||
|
||||
self.task = self.taskgroup.create_task(self.run())
|
||||
|
||||
|
||||
async def run(self):
|
||||
|
||||
if self.metrics:
|
||||
print("RUNNING")
|
||||
self.metrics.state("running")
|
||||
|
||||
while self.running:
|
||||
|
||||
msg = await asyncio.to_thread(self.consumer.receive)
|
||||
|
||||
# expiry = time.time() + self.rate_limit_timeout
|
||||
expiry = time.time() + 10
|
||||
expiry = time.time() + self.rate_limit_timeout
|
||||
|
||||
# This loop is for retry on rate-limit / resource limits
|
||||
while True:
|
||||
|
|
@ -52,8 +63,8 @@ class Consumer:
|
|||
# be retried
|
||||
self.consumer.negative_acknowledge(msg)
|
||||
|
||||
# FIXME
|
||||
# __class__.processing_metric.labels(status="error").inc()
|
||||
if self.metrics:
|
||||
self.metrics.process("error")
|
||||
|
||||
# Break out of retry loop, processes next message
|
||||
break
|
||||
|
|
@ -61,15 +72,22 @@ class Consumer:
|
|||
try:
|
||||
|
||||
print("Handle...")
|
||||
# FIXME
|
||||
# with __class__.request_metric.time():
|
||||
await self.handler(msg, self.consumer)
|
||||
|
||||
if self.metrics:
|
||||
|
||||
with self.metrics.record_time():
|
||||
await self.handler(msg, self.consumer)
|
||||
|
||||
else:
|
||||
await self.handler(msg, self.consumer)
|
||||
|
||||
print("Handled.")
|
||||
|
||||
# Acknowledge successful processing of the message
|
||||
self.consumer.acknowledge(msg)
|
||||
|
||||
# __class__.processing_metric.labels(status="success").inc()
|
||||
if self.metrics:
|
||||
self.metrics.process("success")
|
||||
|
||||
# Break out of retry loop
|
||||
break
|
||||
|
|
@ -79,10 +97,11 @@ class Consumer:
|
|||
print("TooManyRequests: will retry...", flush=True)
|
||||
|
||||
# FIXME
|
||||
# __class__.rate_limit_metric.inc()
|
||||
if self.metrics:
|
||||
self.metrics.rate_limit()
|
||||
|
||||
# Sleep
|
||||
time.sleep(self.rate_limit_retry)
|
||||
await asyncio.sleep(self.rate_limit_retry)
|
||||
|
||||
# Contine from retry loop, just causes a reprocessing
|
||||
continue
|
||||
|
|
@ -95,8 +114,11 @@ class Consumer:
|
|||
# be retried
|
||||
self.consumer.negative_acknowledge(msg)
|
||||
|
||||
# __class__.processing_metric.labels(status="error").inc()
|
||||
if self.metrics:
|
||||
self.metrics.process("error")
|
||||
|
||||
# Break out of retry loop, processes next message
|
||||
break
|
||||
|
||||
if self.metrics:
|
||||
self.metrics.state("stopped")
|
||||
|
|
|
|||
82
trustgraph-base/trustgraph/base/metrics.py
Normal file
82
trustgraph-base/trustgraph/base/metrics.py
Normal file
|
|
@ -0,0 +1,82 @@
|
|||
|
||||
from prometheus_client import start_http_server, Info, Enum, Histogram
|
||||
from prometheus_client import Counter
|
||||
|
||||
class ConsumerMetrics:
|
||||
|
||||
def __init__(self, id, flow=None):
|
||||
|
||||
self.id = id
|
||||
self.flow = flow
|
||||
|
||||
if not hasattr(__class__, "state_metric"):
|
||||
__class__.state_metric = Enum(
|
||||
'consumer_state', 'Consumer state',
|
||||
["id", "flow"],
|
||||
states=['stopped', 'running']
|
||||
)
|
||||
if not hasattr(__class__, "request_metric"):
|
||||
__class__.request_metric = Histogram(
|
||||
'request_latency', 'Request latency (seconds)',
|
||||
["id", "flow"],
|
||||
)
|
||||
if not hasattr(__class__, "processing_metric"):
|
||||
__class__.processing_metric = Counter(
|
||||
'processing_count', 'Processing count',
|
||||
["id", "flow", "status"]
|
||||
)
|
||||
if not hasattr(__class__, "rate_limit_metric"):
|
||||
__class__.rate_limit_metric = Counter(
|
||||
'rate_limit_count', 'Rate limit event count',
|
||||
["id", "flow"]
|
||||
)
|
||||
|
||||
def process(self, status):
|
||||
__class__.processing_metric.labels(
|
||||
id=self.id, flow=self.flow, status=status
|
||||
).inc()
|
||||
|
||||
def rate_limit(self):
|
||||
__class__.rate_limit_metric.labels(
|
||||
id=self.id, flow=self.flow
|
||||
).inc()
|
||||
|
||||
def state(self, state):
|
||||
__class__.state_metric.labels(
|
||||
id=self.id, flow=self.flow
|
||||
).state(state)
|
||||
|
||||
def record_time(self):
|
||||
return __class__.request_metric.labels(
|
||||
id=self.id, flow=self.flow
|
||||
).time()
|
||||
|
||||
class ProducerMetrics:
|
||||
def __init__(self, id, flow=None):
|
||||
|
||||
self.id = id
|
||||
self.flow = flow
|
||||
|
||||
if not hasattr(__class__, "output_metric"):
|
||||
__class__.output_metric = Counter(
|
||||
'output_count', 'Output items created',
|
||||
["id", "flow"]
|
||||
)
|
||||
|
||||
def inc(self):
|
||||
__class__.output_metric.labels(id=self.id, flow=self.flow).inc()
|
||||
|
||||
class ProcessorMetrics:
|
||||
def __init__(self, id):
|
||||
|
||||
self.id = id
|
||||
|
||||
if not hasattr(__class__, "pubsub_metric"):
|
||||
__class__.pubsub_metric = Info(
|
||||
'pubsub', 'Pub/sub configuration',
|
||||
["id"]
|
||||
)
|
||||
|
||||
def info(self, info):
|
||||
__class__.pubsub_metric.labels(id=self.id).info(info)
|
||||
|
||||
|
|
@ -1,18 +1,19 @@
|
|||
|
||||
class Producer:
|
||||
|
||||
def __init__(self, client, queue, schema):
|
||||
def __init__(self, client, queue, schema, metrics=None):
|
||||
self.client = client
|
||||
self.queue = queue
|
||||
self.schema = schema
|
||||
|
||||
self.producer = None
|
||||
self.producer = self.client.publish(self.queue, self.schema)
|
||||
|
||||
self.metrics = metrics
|
||||
|
||||
async def send(self, msg, properties={}):
|
||||
|
||||
if self.producer == None:
|
||||
self.producer = self.client.publish(self.queue, self.schema)
|
||||
|
||||
if self.metrics:
|
||||
self.metrics.inc()
|
||||
|
||||
self.producer.send(msg, properties)
|
||||
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@ from trustgraph.log_level import LogLevel
|
|||
from trustgraph.base import AsyncProcessor, Consumer, Producer
|
||||
|
||||
from . config import Configuration
|
||||
from ... base import ProcessorMetrics, ConsumerMetrics, ProducerMetrics
|
||||
|
||||
module = "config-svc"
|
||||
|
||||
|
|
@ -40,19 +41,13 @@ class Processor(AsyncProcessor):
|
|||
response_schema = ConfigResponse
|
||||
push_schema = ConfigResponse
|
||||
|
||||
self.set_processor_state(id, "starting")
|
||||
|
||||
self.set_pubsub_info(
|
||||
id
|
||||
ProcessorMetrics(id=id).info(
|
||||
{
|
||||
"id": id,
|
||||
"subscriber": subscriber,
|
||||
"request_queue": request_queue,
|
||||
"request_schema": request_schema.__name__,
|
||||
"response_queue": response_queue,
|
||||
"response_schema": request_schema.__name__,
|
||||
# "rate_limit_retry": str(self.rate_limit_retry),
|
||||
# "rate_limit_timeout": str(self.rate_limit_timeout),
|
||||
}
|
||||
)
|
||||
|
||||
|
|
@ -64,25 +59,20 @@ class Processor(AsyncProcessor):
|
|||
}
|
||||
)
|
||||
|
||||
if not hasattr(__class__, "request_metric"):
|
||||
__class__.request_metric = Histogram(
|
||||
'request_latency', 'Request latency (seconds)'
|
||||
)
|
||||
|
||||
if not hasattr(__class__, "processing_metric"):
|
||||
__class__.processing_metric = Counter(
|
||||
'processing_count', 'Processing count',
|
||||
["status"]
|
||||
)
|
||||
self.request_metrics = ConsumerMetrics(id + "-request")
|
||||
self.response_metrics = ProducerMetrics(id + "-response")
|
||||
self.push_metrics = ProducerMetrics(id + "-push")
|
||||
|
||||
self.push_pub = self.publish(
|
||||
queue = push_queue,
|
||||
schema = ConfigPush
|
||||
schema = ConfigPush,
|
||||
metrics = self.push_metrics,
|
||||
)
|
||||
|
||||
self.out_pub = self.publish(
|
||||
self.response_pub = self.publish(
|
||||
queue = response_queue,
|
||||
schema = ConfigResponse
|
||||
schema = ConfigResponse,
|
||||
metrics = self.response_metrics,
|
||||
)
|
||||
|
||||
self.subs = self.subscribe(
|
||||
|
|
@ -90,6 +80,7 @@ class Processor(AsyncProcessor):
|
|||
subscriber = subscriber,
|
||||
schema = request_schema,
|
||||
handler = self.on_message,
|
||||
metrics = self.request_metrics,
|
||||
)
|
||||
|
||||
self.config = Configuration()
|
||||
|
|
@ -130,7 +121,7 @@ class Processor(AsyncProcessor):
|
|||
|
||||
resp = await self.config.handle(v)
|
||||
|
||||
await self.out_pub.send(resp, properties={"id": id})
|
||||
await self.response_pub.send(resp, properties={"id": id})
|
||||
|
||||
consumer.acknowledge(msg)
|
||||
|
||||
|
|
@ -144,7 +135,7 @@ class Processor(AsyncProcessor):
|
|||
text=None,
|
||||
)
|
||||
|
||||
await self.out_pub.send(resp, properties={"id": id})
|
||||
await self.response_pub.send(resp, properties={"id": id})
|
||||
|
||||
consumer.acknowledge(msg)
|
||||
|
||||
|
|
@ -166,7 +157,7 @@ class Processor(AsyncProcessor):
|
|||
)
|
||||
|
||||
parser.add_argument(
|
||||
'-P', '--push-queue',
|
||||
'--push-queue',
|
||||
default=default_push_queue,
|
||||
help=f'Config push queue (default: {default_push_queue})'
|
||||
)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue