Logging Improvements (#14)

* Code changes to retry the failed request on http 429

* Changed the retry logic to try until succeed

* Improved error message

* Change severity from error to warning

* Dropped duplicate logging

* fix: ensure log_offset is properly printing instead of None in log

* chore: bump version to 2.2.3

* Logging improvements for DLS - 1

* Logging improvements for DLS - 2

* Rollback version number

* Fixed the failing test case

* Updated the change log.

* Fixed the code duplication due to mc res

---------

Co-authored-by: Sudharsan-VS <[email protected]>

Author: sfazilud

Diff for: HISTORY.md

@@ -1,5 +1,11 @@
 ## Change Log
 
+### 2.3.0
+
+Logging Improvements
+Bug Fixes
+Improved Error Handling
+
 ### 2.2.2
 
 Security Update: Updated PyYAML in setup.py to 6.0.1, due to a [security vulnerability](https://nvd.nist.gov/vuln/detail/CVE-2020-14343).

Diff for: README.md

@@ -1,4 +1,4 @@
-Duo Log Sync (v2.2.2)
+Duo Log Sync (v2.3.0)
 ===================
 
 [![Issues](https://img.shields.io/github/issues/duosecurity/duo_log_sync)](https://github.com/duosecurity/duo_log_sync/issues)

Diff for: duologsync/__version__.py

@@ -6,7 +6,7 @@
 __title__ = "duologsync"
 __description__ = "Ingest + sync logs from the Duo Admin API into another system."
 __url__ = "https://github.com/duosecurity/duo_log_sync"
-__version__ = "2.2.3"
+__version__ = "2.3.0"
 __author__ = "Duo Security, Inc."
 __license__ = "MIT"
 __email__ = "[email protected]"

Diff for: duologsync/config.py

@@ -255,6 +255,11 @@ def get_value(cls, keys):
 
         return curr_value
 
+    @classmethod
+    def get_config_file_path(cls):
+        """@return the filepath of the config file"""
+        return cls.get_value(['config_file_path'])
+
     @classmethod
     def get_log_filepath(cls):
         """@return the filepath where DLS program messages should be saved"""
@@ -354,6 +359,7 @@ def create_config(cls, config_filepath):
                 if config.get('dls_settings').get('api').get('timeout') < cls.API_TIMEOUT_DEFAULT:
                     config['dls_settings']['api']['timeout'] = cls.API_TIMEOUT_DEFAULT
                     Program.log(f'DuoLogSync: Setting default api timeout to {cls.API_TIMEOUT_DEFAULT} seconds.')
+                config['config_file_path'] = config_filepath
 
         # Will occur when given a bad filepath or a bad file
         except OSError as os_error:

Diff for: duologsync/consumer/consumer.py

@@ -72,11 +72,10 @@ async def consume(self):
                 # Specifically watch out for errno 32 - Broken pipe. This means
                 # that the connect established by writer was reset or shutdown.
                 except BrokenPipeError as broken_pipe_error:
-                    shutdown_reason = f"{broken_pipe_error}"
+                    error_code, error_message = getattr(broken_pipe_error, "args")
+                    shutdown_reason = f"{self.log_type} consumer: [{broken_pipe_error} error_code: {error_code}]"
+                    Program.log(f"{self.log_type} consumer: connection to the destination server was reset or shutdown", logging.ERROR)
                     Program.initiate_shutdown(shutdown_reason)
-                    Program.log(
-                        "DuoLogSync: connection to server was reset", logging.WARNING
-                    )
 
                 finally:
                     if successful_write:
@@ -132,26 +131,16 @@ def update_log_checkpoint(log_type, log_offset, child_account_id):
         @param log_offset   Information to save in the checkpoint file
         """
 
-        Program.log(
-            f"{log_type} consumer: saving latest log offset '{log_offset}' to a checkpointing file",
-            logging.INFO,
-        )
+        checkpoint_filename = f"{log_type}_checkpoint_data_" + child_account_id + ".txt" if child_account_id else f"{log_type}_checkpoint_data.txt"
+        checkpoint_file_path = os.path.join(Config.get_checkpoint_dir(), checkpoint_filename)
 
-        file_path = (
-            os.path.join(
-                Config.get_checkpoint_dir(),
-                f"{log_type}_checkpoint_data_" + child_account_id + ".txt",
-            )
-            if child_account_id
-            else os.path.join(
-                Config.get_checkpoint_dir(), f"{log_type}_checkpoint_data.txt"
-            )
-        )
-
-        checkpoint_filename = file_path
+        if os.path.exists(checkpoint_file_path):
+            Program.log(f"{log_type} consumer: saving latest log offset '{log_offset}' to a checkpoint file '{checkpoint_file_path}'", logging.INFO)
+        else:
+            Program.log(f"{log_type} consumer: checkpoint file '{checkpoint_file_path}' doesn't exist and it will be created to save the latest offset '{log_offset}'", logging.INFO)
 
         # Open file checkpoint_filename in writing mode only
-        checkpoint_file = open(checkpoint_filename, "w")
+        checkpoint_file = open(checkpoint_file_path, "w")
         checkpoint_file.write(json.dumps(log_offset) + "\n")
 
         # According to Python docs, closing a file also flushes the file

Diff for: duologsync/producer/producer.py

@@ -69,14 +69,11 @@ async def produce(self):
                         f"{self.log_type} producer: no new logs available", logging.INFO
                     )
 
-            # Incorrect api_hostname or proxy_server was provided
-            # duo_client throws the same error if either the api_hostname or proxy_server is incorrect
-            except (gaierror, OSError) as error:
-                shutdown_reason = f"{self.log_type} producer: [{error}]"
-                Program.log(
-                    "DuoLogSync: check that the duoclient host and/or proxy_server provided in the config "
-                    "file are correct"
-                )
+            except gaierror as gai_error:
+                shutdown_reason = self.handle_address_info_error(gai_error)
+
+            except OSError as os_error:
+                shutdown_reason = self.handle_os_error(os_error)
 
             # duo_client throws a RuntimeError if the ikey or skey is invalid
             except RuntimeError as runtime_error:
@@ -93,31 +90,48 @@ async def produce(self):
         await self.log_queue.put([])
         Program.log(f"{self.log_type} producer: shutting down", logging.INFO)
 
+    def handle_os_error(self, os_error: OSError):
+        """
+        Handle an OS error gracefully by logging the error and returning a string to indicate that the producer should shut down.
+        """
+        error_code, error_message = getattr(os_error, "args")
+        file_name = getattr(os_error, "filename", None)
+        return f"{self.log_type} producer: [{error_message} error_code: {error_code} file_name: {file_name}]"
+
+    def handle_address_info_error(self, gai_error: gaierror):
+        """
+        Handle an address info error gracefully by logging the error and returning a string to indicate that the producer should shut down.
+        """
+        error_code, error_message = getattr(gai_error, "args")
+        Program.log(f"{self.log_type} producer: make sure that the host information details provider in the configuration file {Config.get_config_file_path()} is correct", logging.ERROR)
+        return f"{self.log_type} producer: [{error_message} error_code: {error_code}]"
+
     def handle_runtime_error_gracefully(self, runtime_error: RuntimeError):
         """
         Handle a runtime error gracefully by checking if the error is eligible for a retry.
         If it is, log the error and return None to indicate that the producer should retry.
         If it is not, log the error and return a string to indicate that the producer should shut down.
         """
-        if self.eligible_for_retry(runtime_error):
-            error_data = getattr(runtime_error, "data", None)
-            error_code = error_data['code'] if error_data else None
+        http_error_code = getattr(runtime_error, "status", None)
+        error_data = getattr(runtime_error, "data", None)
+        error_code = error_data['code'] if error_data else None
+
+        if self.eligible_for_retry(http_error_code):
             Program.log(
-                f"{self.log_type} producer: retrying due to error: {runtime_error} error_code: {error_code}",
+                f"{self.log_type} producer: retrying due to error: {runtime_error} error_code: {error_code} http_status_code: {http_error_code}",
                 logging.WARNING,
             )
 
             return None
 
-        return f"{self.log_type} producer: [{runtime_error}]"
+        return f"{self.log_type} producer: [{runtime_error} error_code: {error_code} http_status_code: {http_error_code}]"
 
     @staticmethod
-    def eligible_for_retry(runtime_error: RuntimeError):
+    def eligible_for_retry(http_error_code):
         """
         Check if the runtime error is eligible for a retry based on the status code.
         See the Config.GRACEFUL_RETRY_STATUS_CODES for the list of status codes that is eligible for a retry.
         """
-        http_error_code = getattr(runtime_error, "status", None)
 
         if http_error_code:
             for http_status_code in Config.GRACEFUL_RETRY_STATUS_CODES:

Diff for: duologsync/util.py

@@ -101,7 +101,7 @@ def get_log_offset(
                 )
             )
             Program.log(
-                f"Recovering log offset from checkpoint file at {checkpoint_file_path}",
+                f"{log_type} producer: recovering log offset from checkpoint file at '{checkpoint_file_path}'",
                 logging.INFO,
             )
 
@@ -111,15 +111,21 @@ def get_log_offset(
                 log_offset = json.loads(checkpoint.read())
 
         # Most likely, the checkpoint file doesn't exist
-        except OSError:
+        except OSError as os_error:
+            error_code, error_message = getattr(os_error, "args")
+            file_name = getattr(os_error, "filename", None)
             display_offset = log_offset
             if log_type in MILLISECOND_BASED_LOG_TYPES:
                 display_offset /= milliseconds_per_second
             iso_timestamp = datetime.fromtimestamp(
                 display_offset, tz=timezone.utc
             ).isoformat()
             Program.log(
-                f"Could not read checkpoint file for {log_type} logs, consuming logs from {iso_timestamp} (UTC)",
+                f"{log_type} producer: could not access checkpoint file '{file_name}' to read log offset due to error: {error_message} error_code: {error_code}",
+                logging.WARNING,
+            )
+            Program.log(
+                f"{log_type} producer: the logs will be consumed from offset: '{iso_timestamp} (UTC)'",
                 logging.INFO,
             )
 

Diff for: duologsync/writer.py

@@ -7,6 +7,8 @@
 import logging
 import socket
 from socket import gaierror
+
+from duologsync.config import Config
 from duologsync.program import Program
 
 
@@ -113,9 +115,7 @@ async def create_writer(self, host, port, cert_filepath):
                     logging.INFO)
 
         # Message to be logged if an error occurs in this function
-        help_message = (
-            f"DuoLogSync: check that host-{host} and port-{port} "
-            "are correct in the config file")
+        help_message = f"DuoLogSync: check that host-{host} and port-{port} are correct in the configuration file '{Config.get_config_file_path()}'"
         writer = None
 
         try:
@@ -132,20 +132,20 @@ async def create_writer(self, host, port, cert_filepath):
                 writer = await Writer.create_tcp_writer(host, port)
 
         # Failed to open the certificate file
-        except FileNotFoundError:
-            shutdown_reason = f"{cert_filepath} could not be opened."
-            help_message = (
-                'DuoLogSync: Make sure the filepath for SSL cert file is '
-                'correct.')
+        except FileNotFoundError as fnf_error:
+            error_code, error_message = getattr(fnf_error, "args")
+            shutdown_reason = f"certificate file '{cert_filepath}' could not be opened due to error: {error_message} error_code: {error_code}"
+            help_message = f"make sure that certificate file '{cert_filepath}' to establish SSL connection is correct."
 
         # Couldn't establish a connection within 60 seconds
         except asyncio.TimeoutError:
-            shutdown_reason = 'connection to server timed-out after 60 seconds'
+            shutdown_reason = 'connection to the destination server timed-out after 60 seconds'
 
         # If an invalid hostname or port number is given or simply failed to
         # connect using the host and port given
         except (gaierror, OSError) as error:
-            shutdown_reason = f"{error}"
+            error_code, error_message = getattr(error, "args")
+            shutdown_reason = f"error while connecting to {host}:{port} error_message: {error_message} error_code: {error_code}"
 
         # An error did not occur and the writer was successfully created
         else:

Diff for: tests/test_config.py

@@ -146,6 +146,7 @@ def test_get_account_block_list(self):
     def test_create_config_normal(self):
         config_filepath = 'tests/resources/config_files/standard.yml'
         correct_config = {
+            'config_file_path': config_filepath,
             'version': '1.0.0',
             'dls_settings': {
                 'log_filepath': '/tmp/duologsync.log',