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

Author: JuanFranDevo

CHANGELOG.md

@@ -4,6 +4,11 @@ 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).
 
+## [3.6.5] - 2023-01-27
+### 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.
+
+
 ## [3.6.4] - 2022-07-21
 ### Fixed
  * Fixed bug when processing keep alive empty tokens

devo/__version__.py

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

devo/sender/data.py

@@ -286,10 +286,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("Problems with args passed to Sender")
@@ -298,7 +300,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 = {}
 
@@ -344,6 +348,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(
@@ -407,6 +412,7 @@ def __connect_ssl(self):
                 raise ssl.SSLError
 
             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'
@@ -511,6 +517,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):
@@ -553,6 +566,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("Send error")
             sent += len(part)
@@ -577,6 +591,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("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
 

requirements.txt

@@ -2,5 +2,5 @@ click==8.1.3
 PyYAML==6.0
 requests==2.27.1
 pem==21.2.0
-pyopenssl==22.0.0
+pyopenssl==22.1.*
 urllib3>=1.26.5