mirror of
https://github.com/trustgraph-ai/trustgraph.git
synced 2026-05-01 03:16:23 +02:00
119 lines
2.8 KiB
Python
119 lines
2.8 KiB
Python
|
|
import os
|
|
import argparse
|
|
import pulsar
|
|
import _pulsar
|
|
import time
|
|
from prometheus_client import start_http_server, Info
|
|
|
|
from .. log_level import LogLevel
|
|
|
|
class BaseProcessor:
|
|
|
|
default_pulsar_host = os.getenv("PULSAR_HOST", 'pulsar://pulsar:6650')
|
|
|
|
def __init__(self, **params):
|
|
|
|
self.client = None
|
|
|
|
if not hasattr(__class__, "params_metric"):
|
|
__class__.params_metric = Info(
|
|
'params', 'Parameters configuration'
|
|
)
|
|
|
|
# FIXME: Maybe outputs information it should not
|
|
__class__.params_metric.info({
|
|
k: str(params[k])
|
|
for k in params
|
|
})
|
|
|
|
pulsar_host = params.get("pulsar_host", self.default_pulsar_host)
|
|
log_level = params.get("log_level", LogLevel.INFO)
|
|
|
|
self.pulsar_host = pulsar_host
|
|
|
|
self.client = pulsar.Client(
|
|
pulsar_host,
|
|
logger=pulsar.ConsoleLogger(log_level.to_pulsar())
|
|
)
|
|
|
|
def __del__(self):
|
|
|
|
if self.client:
|
|
self.client.close()
|
|
|
|
@staticmethod
|
|
def add_args(parser):
|
|
|
|
parser.add_argument(
|
|
'-p', '--pulsar-host',
|
|
default=__class__.default_pulsar_host,
|
|
help=f'Pulsar host (default: {__class__.default_pulsar_host})',
|
|
)
|
|
|
|
parser.add_argument(
|
|
'-l', '--log-level',
|
|
type=LogLevel,
|
|
default=LogLevel.INFO,
|
|
choices=list(LogLevel),
|
|
help=f'Output queue (default: info)'
|
|
)
|
|
|
|
parser.add_argument(
|
|
'--metrics',
|
|
action=argparse.BooleanOptionalAction,
|
|
default=True,
|
|
help=f'Metrics enabled (default: true)',
|
|
)
|
|
|
|
parser.add_argument(
|
|
'-P', '--metrics-port',
|
|
type=int,
|
|
default=8000,
|
|
help=f'Pulsar host (default: 8000)',
|
|
)
|
|
|
|
def run(self):
|
|
raise RuntimeError("Something should have implemented the run method")
|
|
|
|
@classmethod
|
|
def start(cls, prog, doc):
|
|
|
|
parser = argparse.ArgumentParser(
|
|
prog=prog,
|
|
description=doc
|
|
)
|
|
|
|
cls.add_args(parser)
|
|
|
|
args = parser.parse_args()
|
|
args = vars(args)
|
|
|
|
print(args)
|
|
|
|
if args["metrics"]:
|
|
start_http_server(args["metrics_port"])
|
|
|
|
while True:
|
|
|
|
try:
|
|
|
|
p = cls(**args)
|
|
p.run()
|
|
|
|
except KeyboardInterrupt:
|
|
print("Keyboard interrupt.")
|
|
return
|
|
|
|
except _pulsar.Interrupted:
|
|
print("Pulsar Interrupted.")
|
|
return
|
|
|
|
except Exception as e:
|
|
|
|
print(type(e))
|
|
|
|
print("Exception:", e, flush=True)
|
|
print("Will retry...", flush=True)
|
|
|
|
time.sleep(10)
|