Add event logs

portality/events/kafka_consumer.py

@@ -3,6 +3,7 @@
 
 from portality.app import app as doajapp
 from portality.bll import DOAJ
+from portality import util
 from portality.models import Event
 
 broker = doajapp.config.get("KAFKA_BROKER")
@@ -12,6 +13,7 @@
 topic = app.topic(topic_name)
 
 event_counter = 0
+event_logger = util.custom_timed_rotating_logger('consumer_log.log')
 
 
 @app.agent(topic)
@@ -20,6 +22,7 @@ async def handle_event(stream):
     with doajapp.test_request_context("/"):
         svc = DOAJ.eventsService()
         async for event in stream:
+            event_logger.info(event)
             event_counter += 1
             doajapp.logger.info(f"Kafka event count {event_counter}")
             # TODO uncomment the following line once the Event model is fixed to Kafka

portality/events/kafka_producer.py

@@ -1,7 +1,7 @@
 import json
 import kafka
 from portality.core import app
-from portality import app_email
+from portality import app_email, util
 from portality.events.shortcircuit import send_event as shortcircuit_send_event
 from portality.events.system_status_check import KafkaStatusCheck
 
@@ -18,6 +18,7 @@ def handle_exception(error_msg, exception):
     )
 
 producer = None
+event_logger = util.custom_timed_rotating_logger('producer_log.log')
 
 def kafka_producer():
     global producer
@@ -38,6 +39,7 @@ def kafka_producer():
 
 def send_event(event):
     try:
+        event_logger.info(event.data)
         if kafka_status and kafka_status.is_active() and kafka_producer():
             future = producer.send('events', value=event.serialise())
             future.get(timeout=60)

portality/events/shortcircuit.py

@@ -1,6 +1,9 @@
 from portality.bll import DOAJ
+from portality import util
 
+event_logger = util.custom_timed_rotating_logger('shortcircuit_log.log')
 
 def send_event(event):
+    event_logger.info(event.data)
     svc = DOAJ.eventsService()
     svc.consume(event)

portality/util.py

@@ -186,3 +186,30 @@ def get_full_url_safe(endpoint):
     except werkzeug.routing.BuildError:
         app.logger.warning(f'endpoint not found -- [{endpoint}]')
         return None
+
+def custom_timed_rotating_logger(file_name):
+    """Custom Logger to log to specified file name"""
+    import os
+    import logging
+    from logging.handlers import TimedRotatingFileHandler
+    # Create a logger
+    logger = logging.getLogger(os.path.splitext(file_name)[0])
+    logger.setLevel(logging.DEBUG)  # Set the logging level
+
+    # Get the user's home directory
+    user_home = os.path.expanduser("~")
+    log_dir = os.path.join(user_home, 'doaj_logs')
+    if not os.path.exists(log_dir):
+        os.makedirs(log_dir)
+    log_filename = os.path.join(log_dir, file_name)
+    # Rotate every day. Keep 30 days worth of backup.
+    handler = TimedRotatingFileHandler(log_filename, when="D", interval=1, backupCount=30)
+
+    # Create a formatter and add it to the handler
+    formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
+    handler.setFormatter(formatter)
+
+    # Add the handler to the logger
+    logger.addHandler(handler)
+
+    return logger