Fix/par 17098 (#161)

* fix(PAR-17098): Devo Python SDK causes log messages to be dropped

* fix(PAR-17098): Devo Python SDK causes log messages to be dropped

Author: JuanFranDevo

CHANGELOG.md

@@ -4,6 +4,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
+## [5.0.3] - 2023-01-26
+### Fixed
+ * Ingestion endpoint has an inactivity timeout that when reached closes the connection. `devo-sdk` is aware of such a timeout and restart connection before is reached. New parameter `inactivity_timeout` in class `Sender` to set up it. Its default value is 30 seconds.
+ * Syntax error when calling `Path.is_file()`
+
+
 ## [5.0.2] - 2023-01-04
 ### Fixed
  * `pyopenssl` dependency bumped

devo/__version__.py

@@ -1,6 +1,6 @@
 __description__ = 'Devo Python Library.'
 __url__ = 'http://www.devo.com'
-__version__ = "5.0.2"
+__version__ = "5.0.3"
 __author__ = 'Devo'
 __author_email__ = 'support@devo.com'
 __license__ = 'MIT'

devo/sender/data.py

@@ -125,7 +125,7 @@ def check_config_files_path(self):
                 if message.errno == errno.EACCES:
                     raise DevoSenderException(
                         ERROR_MSGS.CANT_READ_CONFIG_FILE % (
-                        file, str(message))) \
+                            file, str(message))) \
                         from message
                 else:
                     raise DevoSenderException(
@@ -303,11 +303,12 @@ class Sender(logging.Handler):
     :param con_type: TCP or SSL, default SSL, you can pass it in
     config object too
     :param timeout: timeout for socket
+    :param inactivity_timeout: inactivity timeout for Ingestion balancer, so connection is restarted before reaching
     :param debug: For more info in console/logger output
     :param logger: logger. Default sys.console
     """
 
-    def __init__(self, config=None, con_type=None,
+    def __init__(self, config=None, con_type=None, inactivity_timeout=30,
                  timeout=30, debug=False, logger=None):
         if config is None:
             raise DevoSenderException(ERROR_MSGS.PROBLEMS_WITH_SENDER_ARGS)
@@ -316,7 +317,9 @@ def __init__(self, config=None, con_type=None,
         self.reconnection = 0
         self.debug = debug
         self.socket_timeout = timeout
+        self.inactivity_timeout = inactivity_timeout
         self.socket_max_connection = 3600 * 1000
+        self.last_message = int(time.time())
         self.buffer = SenderBuffer()
         self.logging = {}
 
@@ -362,6 +365,7 @@ def __connect_tcp_socket(self):
         self.socket.settimeout(self.socket_timeout)
         try:
             self.socket.connect(self._sender_config.address)
+            self.last_message = int(time.time())
         except socket.error as error:
             self.close()
             raise DevoSenderException(
@@ -392,38 +396,36 @@ def __connect_ssl(self):
             raise DevoSenderException(
                 ERROR_MSGS.PFX_CERTIFICATE_READ_FAILED % str(error)) from error
         try:
-            try:
-                if self._sender_config.key is not None \
-                        and self._sender_config.chain is not None \
-                        and self._sender_config.cert is not None:
-
-                    context = ssl.create_default_context(
-                        cafile=self._sender_config.chain)
-
-                    if self._sender_config.sec_level is not None:
-                        context.set_ciphers(
-                            "DEFAULT@SECLEVEL={!s}"
-                            .format(self._sender_config.sec_level))
-
-                    context.check_hostname = self._sender_config.check_hostname
-
-                    if self._sender_config.verify_mode is not None:
-                        context.verify_mode = self._sender_config.verify_mode
-
-                    context.load_cert_chain(keyfile=self._sender_config.key,
-                                            certfile=self._sender_config.cert)
-                    self.socket = \
-                        context.wrap_socket(
-                            self.socket,
-                            server_hostname=self._sender_config.address[0]
-                        )
-                else:
-                    self.socket = ssl.wrap_socket(self.socket,
-                                                  cert_reqs=ssl.CERT_NONE)
-            except ssl.SSLError:
-                raise ssl.SSLError
+            if self._sender_config.key is not None \
+                    and self._sender_config.chain is not None \
+                    and self._sender_config.cert is not None:
+
+                context = ssl.create_default_context(
+                    cafile=self._sender_config.chain)
+
+                if self._sender_config.sec_level is not None:
+                    context.set_ciphers(
+                        "DEFAULT@SECLEVEL={!s}"
+                        .format(self._sender_config.sec_level))
+
+                context.check_hostname = self._sender_config.check_hostname
+
+                if self._sender_config.verify_mode is not None:
+                    context.verify_mode = self._sender_config.verify_mode
+
+                context.load_cert_chain(keyfile=self._sender_config.key,
+                                        certfile=self._sender_config.cert)
+                self.socket = \
+                    context.wrap_socket(
+                        self.socket,
+                        server_hostname=self._sender_config.address[0]
+                    )
+            else:
+                self.socket = ssl.wrap_socket(self.socket,
+                                              cert_reqs=ssl.CERT_NONE)
 
             self.socket.connect(self._sender_config.address)
+            self.last_message = int(time.time())
             self.reconnection += 1
             if self.debug:
                 self.logger.debug('Conected to %s|%s'
@@ -527,6 +529,13 @@ def __status(self):
         if self.socket_max_connection < timeit:
             self.close()
             return False
+
+        # If there is no activity (connection or message sent) for an amount of time bigger then the inactivity
+        # timeout, the balancer may have already close the connection. Close it and reconnect.
+        if int(time.time()) - self.last_message > self.inactivity_timeout:
+            self.close()
+            return False
+
         return True
 
     def close(self):
@@ -569,6 +578,7 @@ def __send_oc(self, record):
         for iteration in range(0, total):
             part = record[int(iteration * 4096):
                           int((iteration + 1) * 4096)]
+            self.last_message = int(time.time())
             if self.socket.sendall(part) is not None:
                 raise DevoSenderException(ERROR_MSGS.SEND_ERROR)
             sent += len(part)
@@ -593,6 +603,7 @@ def send_raw(self, record, multiline=False, zip=False):
                     if not multiline and not zip:
                         msg = self.__encode_record(record)
                         sent = len(msg)
+                        self.last_message = int(time.time())
                         if self.socket.sendall(msg) is not None:
                             raise DevoSenderException(ERROR_MSGS.SEND_ERROR)
                         return 1

docs/sender/data.md

@@ -14,6 +14,7 @@ Variable descriptions
 + config **(_SenderConfigSSL_, _SenderConfigTCP_ or _dict_)**: address, port, keypath, chainpath, etc
 + con_type **(_string_)**: TCP or SSL, default SSL, you can pass it in config object too
 + timeout **(_int_)**: timeout for socket
++ inactivity_timeout **(_int_)**: inactivity timeout for Ingestion balancer, so connection is restarted before reaching
 + debug **(_bool_)**: True or False, for show more info in console/logger output
 + logger **(_string_)**: logger. Default sys.console