feat: add command to debug sync events (#2190)

This commit is contained in:
Carlos Quintana 2024-08-21 12:35:08 +02:00 committed by GitHub
parent 33c418d7c6
commit 57991f4d6b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 96 additions and 17 deletions

View File

@ -4,6 +4,7 @@ from sys import argv, exit
from app.config import EVENT_LISTENER_DB_URI from app.config import EVENT_LISTENER_DB_URI
from app.log import LOG from app.log import LOG
from events import event_debugger
from events.runner import Runner from events.runner import Runner
from events.event_source import DeadLetterEventSource, PostgresEventSource from events.event_source import DeadLetterEventSource, PostgresEventSource
from events.event_sink import ConsoleEventSink, HttpEventSink from events.event_sink import ConsoleEventSink, HttpEventSink
@ -46,32 +47,67 @@ def main(mode: Mode, dry_run: bool, max_retries: int):
runner.run() runner.run()
def debug_event(event_id: str):
LOG.i(f"Debugging event {event_id}")
try:
event_id_int = int(event_id)
except ValueError:
raise ValueError(f"Invalid event id: {event_id}")
event_debugger.debug_event(event_id_int)
def run_event(event_id: str, delete_on_success: bool):
LOG.i(f"Running event {event_id}")
try:
event_id_int = int(event_id)
except ValueError:
raise ValueError(f"Invalid event id: {event_id}")
event_debugger.run_event(event_id_int, delete_on_success)
def args(): def args():
parser = argparse.ArgumentParser(description="Run event listener") parser = argparse.ArgumentParser(description="Run event listener")
parser.add_argument( subparsers = parser.add_subparsers(dest="command")
"mode",
help="Mode to run", listener_parser = subparsers.add_parser(Mode.LISTENER.value)
choices=[Mode.DEAD_LETTER.value, Mode.LISTENER.value], listener_parser.add_argument(
"--max-retries", type=int, default=_DEFAULT_MAX_RETRIES
) )
parser.add_argument( listener_parser.add_argument("--dry-run", action="store_true")
"max_retries",
help="Max retries to consider an event as error and not try to process it again", dead_letter_parser = subparsers.add_parser(Mode.DEAD_LETTER.value)
type=int, dead_letter_parser.add_argument(
nargs="?", "--max-retries", type=int, default=_DEFAULT_MAX_RETRIES
default=_DEFAULT_MAX_RETRIES,
) )
parser.add_argument("--dry-run", help="Dry run mode", action="store_true") dead_letter_parser.add_argument("--dry-run", action="store_true")
debug_parser = subparsers.add_parser("debug")
debug_parser.add_argument("event_id", help="ID of the event to debug")
run_parser = subparsers.add_parser("run")
run_parser.add_argument("event_id", help="ID of the event to run")
run_parser.add_argument("--delete-on-success", action="store_true")
return parser.parse_args() return parser.parse_args()
if __name__ == "__main__": if __name__ == "__main__":
if len(argv) < 2: if len(argv) < 2:
print("Invalid usage. Pass 'listener' or 'dead_letter' as argument") print("Invalid usage. Pass a valid subcommand as argument")
exit(1) exit(1)
args = args() args = args()
main(
mode=Mode.from_str(args.mode), if args.command in [Mode.LISTENER.value, Mode.DEAD_LETTER.value]:
dry_run=args.dry_run, main(
max_retries=args.max_retries, mode=Mode.from_str(args.command),
) dry_run=args.dry_run,
max_retries=args.max_retries,
)
elif args.command == "debug":
debug_event(args.event_id)
elif args.command == "run":
run_event(args.event_id, args.delete_on_success)
else:
print("Invalid command")
exit(1)

43
events/event_debugger.py Normal file
View File

@ -0,0 +1,43 @@
from app.events.generated import event_pb2
from app.models import SyncEvent
from events.event_sink import HttpEventSink
def debug_event(event_id: int):
event = SyncEvent.get_by(id=event_id)
if not event:
print("Event not found")
return
print(f"Info for event {event_id}")
print(f"- Created at: {event.created_at}")
print(f"- Updated at: {event.updated_at}")
print(f"- Taken time: {event.taken_time}")
print(f"- Retry count: {event.retry_count}")
print()
print("Event contents")
event_contents = event.content
parsed = event_pb2.Event.FromString(event_contents)
print(f"- UserID: {parsed.user_id}")
print(f"- ExternalUserID: {parsed.external_user_id}")
print(f"- PartnerID: {parsed.partner_id}")
content = parsed.content
print(f"Content: {content}")
def run_event(event_id: int, delete_on_success: bool = True):
event = SyncEvent.get_by(id=event_id)
if not event:
print("Event not found")
return
print(f"Processing event {event_id}")
sink = HttpEventSink()
res = sink.process(event)
if res:
print(f"Processed event {event_id}")
if delete_on_success:
SyncEvent.delete(event_id, commit=True)