mirror of
https://github.com/simple-login/app.git
synced 2024-11-10 21:27:10 +01:00
6862ed3602
* fix: commit transaction after taking event * feat: allow to reconnect to postgres for event listener * chore: log sync events pending to process to metrics * fix: make dead_letter runner able to process events without needing to have lock on the event * chore: close Session after reconnect * refactor: make EventSource emit only events that can be processed
43 lines
1.5 KiB
Python
43 lines
1.5 KiB
Python
import arrow
|
|
import newrelic.agent
|
|
|
|
from app.log import LOG
|
|
from app.models import SyncEvent
|
|
from events.event_sink import EventSink
|
|
from events.event_source import EventSource
|
|
|
|
|
|
class Runner:
|
|
def __init__(self, source: EventSource, sink: EventSink):
|
|
self.__source = source
|
|
self.__sink = sink
|
|
|
|
def run(self):
|
|
self.__source.run(self.__on_event)
|
|
|
|
@newrelic.agent.background_task()
|
|
def __on_event(self, event: SyncEvent):
|
|
try:
|
|
event_created_at = event.created_at
|
|
start_time = arrow.now()
|
|
success = self.__sink.process(event)
|
|
if success:
|
|
event_id = event.id
|
|
SyncEvent.delete(event.id, commit=True)
|
|
LOG.info(f"Marked {event_id} as done")
|
|
|
|
end_time = arrow.now() - start_time
|
|
time_between_taken_and_created = start_time - event_created_at
|
|
|
|
newrelic.agent.record_custom_metric("Custom/sync_event_processed", 1)
|
|
newrelic.agent.record_custom_metric(
|
|
"Custom/sync_event_process_time", end_time.total_seconds()
|
|
)
|
|
newrelic.agent.record_custom_metric(
|
|
"Custom/sync_event_elapsed_time",
|
|
time_between_taken_and_created.total_seconds(),
|
|
)
|
|
except Exception as e:
|
|
LOG.warn(f"Exception processing event [id={event.id}]: {e}")
|
|
newrelic.agent.record_custom_metric("Custom/sync_event_failed", 1)
|