Adapt code to elasticsearch 8.x version

Elasticsearch introduced braking change by:
- removing RequestsHttpConnection
- removing use_ssl from Elasticsearch class
SSL is not set as 'scheme' string (http/https) in host dict.
Custom authorization need to set node_class to requests.

Author: thoscilo

Diff for: README.rst

@@ -42,20 +42,20 @@ This library requires the following dependencies
 
 Additional requirements for Kerberos support
 ============================================
-Additionally, the package support optionally kerberos authentication by adding the following dependecy
+Additionally, the package support optionally kerberos authentication by adding the following dependency
  - requests-kerberos
 
 Additional requirements for AWS IAM user authentication (request signing)
 =========================================================================
-Additionally, the package support optionally AWS IAM user authentication by adding the following dependecy
+Additionally, the package support optionally AWS IAM user authentication by adding the following dependency
  - requests-aws4auth
 
 Using the handler in  your program
 ==================================
 To initialise and create the handler, just add the handler to your logger as follow ::
 
     from cmreslogging.handlers import CMRESHandler
-    handler = CMRESHandler(hosts=[{'host': 'localhost', 'port': 9200}],
+    handler = CMRESHandler(hosts=[{'host': 'localhost', 'port': 9200, 'scheme': 'http'}],
                                auth_type=CMRESHandler.AuthType.NO_AUTH,
                                es_index_name="my_python_index")
     log = logging.getLogger("PythonTest")
@@ -65,7 +65,7 @@ To initialise and create the handler, just add the handler to your logger as fol
 You can add fields upon initialisation, providing more data of the execution context ::
 
     from cmreslogging.handlers import CMRESHandler
-    handler = CMRESHandler(hosts=[{'host': 'localhost', 'port': 9200}],
+    handler = CMRESHandler(hosts=[{'host': 'localhost', 'port': 9200, 'scheme': 'http'}],
                                auth_type=CMRESHandler.AuthType.NO_AUTH,
                                es_index_name="my_python_index",
                                es_additional_fields={'App': 'MyAppName', 'Environment': 'Dev'})
@@ -95,17 +95,18 @@ Kibana on top of elasticsearch
 Initialisation parameters
 =========================
 The constructors takes the following parameters:
- - hosts:  The list of hosts that elasticsearch clients will connect, multiple hosts are allowed, for example ::
+  - hosts:  The list of hosts that elasticsearch clients will connect, multiple hosts are allowed.
+   Use ```'scheme'``` to determinate if use SSL (`use_ssl` is deprecated). To use SSL set ```'scheme': 'https'```, or if you don't need SSL Sset ```'scheme': 'http'```.
+   for example::
 
-    [{'host':'host1','port':9200}, {'host':'host2','port':9200}]
+    [{'host':'host1','port':9200, 'scheme': 'https'}, {'host':'host2','port':9200, 'scheme': 'http'}]
 
 
  - auth_type: The authentication currently support CMRESHandler.AuthType = NO_AUTH, BASIC_AUTH, KERBEROS_AUTH
  - auth_details: When CMRESHandler.AuthType.BASIC_AUTH is used this argument must contain a tuple of string with the user and password that will be used to authenticate against the Elasticsearch servers, for example ('User','Password')
  - aws_access_key: When ``CMRESHandler.AuthType.AWS_SIGNED_AUTH`` is used this argument must contain the AWS key id of the  the AWS IAM user
  - aws_secret_key: When ``CMRESHandler.AuthType.AWS_SIGNED_AUTH`` is used this argument must contain the AWS secret key of the  the AWS IAM user
  - aws_region: When ``CMRESHandler.AuthType.AWS_SIGNED_AUTH`` is used this argument must contain the AWS region of the  the AWS Elasticsearch servers, for example ``'us-east'``
- - use_ssl: A boolean that defines if the communications should use SSL encrypted communication
  - verify_ssl: A boolean that defines if the SSL certificates are validated or not
  - buffer_size: An int, Once this size is reached on the internal buffer results are flushed into ES
  - flush_frequency_in_sec: A float representing how often and when the buffer will be flushed
@@ -139,7 +140,7 @@ they can be plotted on Kibana, or the SQL statements that Django executed. ::
             'elasticsearch': {
                 'level': 'DEBUG',
                 'class': 'cmreslogging.handlers.CMRESHandler',
-                'hosts': [{'host': 'localhost', 'port': 9200}],
+                'hosts': [{'host': 'localhost', 'port': 9200, 'scheme': 'http'}],
                 'es_index_name': 'my_python_app',
                 'es_additional_fields': {'App': 'Test', 'Environment': 'Dev'},
                 'auth_type': CMRESHandler.AuthType.NO_AUTH,

Diff for: cmreslogging/handlers.py

@@ -7,7 +7,7 @@
 from threading import Timer, Lock
 from enum import Enum
 from elasticsearch import helpers as eshelpers
-from elasticsearch import Elasticsearch, RequestsHttpConnection
+from elasticsearch import Elasticsearch
 
 try:
     from requests_kerberos import HTTPKerberosAuth, DISABLED
@@ -58,13 +58,12 @@ class IndexNameFrequency(Enum):
         YEARLY = 3
 
     # Defaults for the class
-    __DEFAULT_ELASTICSEARCH_HOST = [{'host': 'localhost', 'port': 9200}]
+    __DEFAULT_ELASTICSEARCH_HOST = [{'host': 'localhost', 'port': 9200, 'scheme': 'http'}]
     __DEFAULT_AUTH_USER = ''
     __DEFAULT_AUTH_PASSWD = ''
     __DEFAULT_AWS_ACCESS_KEY = ''
     __DEFAULT_AWS_SECRET_KEY = ''
     __DEFAULT_AWS_REGION = ''
-    __DEFAULT_USE_SSL = False
     __DEFAULT_VERIFY_SSL = True
     __DEFAULT_AUTH_TYPE = AuthType.NO_AUTH
     __DEFAULT_INDEX_FREQUENCY = IndexNameFrequency.DAILY
@@ -129,7 +128,6 @@ def __init__(self,
                  aws_secret_key=__DEFAULT_AWS_SECRET_KEY,
                  aws_region=__DEFAULT_AWS_REGION,
                  auth_type=__DEFAULT_AUTH_TYPE,
-                 use_ssl=__DEFAULT_USE_SSL,
                  verify_ssl=__DEFAULT_VERIFY_SSL,
                  buffer_size=__DEFAULT_BUFFER_SIZE,
                  flush_frequency_in_sec=__DEFAULT_FLUSH_FREQ_INSEC,
@@ -142,11 +140,13 @@ def __init__(self,
         """ Handler constructor
 
         :param hosts: The list of hosts that elasticsearch clients will connect. The list can be provided
-                    in the format ```[{'host':'host1','port':9200}, {'host':'host2','port':9200}]``` to
-                    make sure the client supports failover of one of the instertion nodes
-        :param auth_details: When ```CMRESHandler.AuthType.BASIC_AUTH``` is used this argument must contain
-                    a tuple of string with the user and password that will be used to authenticate against
-                    the Elasticsearch servers, for example```('User','Password')
+                    in the format ```[{'host':'host1','port':9200, 'scheme': 'http'},
+                    {'host':'host2','port':9200, 'scheme': 'https'}]``` to make sure the client supports
+                    failover of one of the insertion nodes
+        :param auth_details: When ```CMRESHandler.AuthType.BASIC_AUTH``` or ```CMRESHandler.AuthType.NTLM_AUTH```
+                    is used this argument must contain a tuple of string with the user and password
+                    that will be used to authenticate against the Elasticsearch servers,
+                    for example```('User','Password')
         :param aws_access_key: When ```CMRESHandler.AuthType.AWS_SIGNED_AUTH``` is used this argument must contain
                     the AWS key id of the  the AWS IAM user
         :param aws_secret_key: When ```CMRESHandler.AuthType.AWS_SIGNED_AUTH``` is used this argument must contain
@@ -155,7 +155,6 @@ def __init__(self,
                     the AWS region of the  the AWS Elasticsearch servers, for example```'us-east'
         :param auth_type: The authentication type to be used in the connection ```CMRESHandler.AuthType```
                     Currently, NO_AUTH, BASIC_AUTH, KERBEROS_AUTH are supported
-        :param use_ssl: A boolean that defines if the communications should use SSL encrypted communication
         :param verify_ssl: A boolean that defines if the SSL certificates are validated or not
         :param buffer_size: An int, Once this size is reached on the internal buffer results are flushed into ES
         :param flush_frequency_in_sec: A float representing how often and when the buffer will be flushed, even
@@ -182,7 +181,6 @@ def __init__(self,
         self.aws_secret_key = aws_secret_key
         self.aws_region = aws_region
         self.auth_type = auth_type
-        self.use_ssl = use_ssl
         self.verify_certs = verify_ssl
         self.buffer_size = buffer_size
         self.flush_frequency_in_sec = flush_frequency_in_sec
@@ -212,19 +210,15 @@ def __get_es_client(self):
         if self.auth_type == CMRESHandler.AuthType.NO_AUTH:
             if self._client is None:
                 self._client = Elasticsearch(hosts=self.hosts,
-                                             use_ssl=self.use_ssl,
                                              verify_certs=self.verify_certs,
-                                             connection_class=RequestsHttpConnection,
                                              serializer=self.serializer)
             return self._client
 
         if self.auth_type == CMRESHandler.AuthType.BASIC_AUTH:
             if self._client is None:
                 return Elasticsearch(hosts=self.hosts,
                                      http_auth=self.auth_details,
-                                     use_ssl=self.use_ssl,
                                      verify_certs=self.verify_certs,
-                                     connection_class=RequestsHttpConnection,
                                      serializer=self.serializer)
             return self._client
 
@@ -233,11 +227,10 @@ def __get_es_client(self):
                 raise EnvironmentError("Kerberos module not available. Please install \"requests-kerberos\"")
             # For kerberos we return a new client each time to make sure the tokens are up to date
             return Elasticsearch(hosts=self.hosts,
-                                 use_ssl=self.use_ssl,
                                  verify_certs=self.verify_certs,
-                                 connection_class=RequestsHttpConnection,
                                  http_auth=HTTPKerberosAuth(mutual_authentication=DISABLED),
-                                 serializer=self.serializer)
+                                 serializer=self.serializer,
+                                 node_class='requests')
 
         if self.auth_type == CMRESHandler.AuthType.AWS_SIGNED_AUTH:
             if not AWS4AUTH_SUPPORTED:
@@ -247,9 +240,7 @@ def __get_es_client(self):
                 self._client = Elasticsearch(
                     hosts=self.hosts,
                     http_auth=awsauth,
-                    use_ssl=self.use_ssl,
                     verify_certs=True,
-                    connection_class=RequestsHttpConnection,
                     serializer=self.serializer
                 )
             return self._client

Diff for: tests/test_cmreshandler.py

@@ -10,16 +10,21 @@
 class CMRESHandlerTestCase(unittest.TestCase):
     DEFAULT_ES_SERVER = 'localhost'
     DEFAULT_ES_PORT = 9200
+    DEFAULT_ES_SSL_SCHEME = 'http'
 
     def getESHost(self):
-        return os.getenv('TEST_ES_SERVER',CMRESHandlerTestCase.DEFAULT_ES_SERVER)
+        return os.getenv('TEST_ES_SERVER', CMRESHandlerTestCase.DEFAULT_ES_SERVER)
 
     def getESPort(self):
         try:
-            return int(os.getenv('TEST_ES_PORT',CMRESHandlerTestCase.DEFAULT_ES_PORT))
+            return int(os.getenv('TEST_ES_PORT', CMRESHandlerTestCase.DEFAULT_ES_PORT))
         except ValueError:
             return CMRESHandlerTestCase.DEFAULT_ES_PORT
 
+    @staticmethod
+    def get_ES_scheme():
+        return os.getenv('TEST_ES_SSL_SCHEME', CMRESHandlerTestCase.DEFAULT_ES_SSL_SCHEME)
+
     def setUp(self):
         self.log = logging.getLogger("MyTestCase")
         test_handler = logging.StreamHandler(stream=sys.stderr)
@@ -29,10 +34,11 @@ def tearDown(self):
         del self.log
 
     def test_ping(self):
-        handler = CMRESHandler(hosts=[{'host': self.getESHost(), 'port': self.getESPort()}],
                                auth_type=CMRESHandler.AuthType.NO_AUTH,
+        handler = CMRESHandler(hosts=[{'host': self.getESHost(),
+                                       'port': self.getESPort(),
+                                       'scheme': self.get_ES_scheme()}],
                                es_index_name="pythontest",
-                               use_ssl=False,
                                raise_on_indexing_exceptions=True)
         es_test_server_is_up = handler.test_es_source()
         self.assertEqual(True, es_test_server_is_up)
@@ -41,6 +47,9 @@ def test_buffered_log_insertion_flushed_when_buffer_full(self):
         handler = CMRESHandler(hosts=[{'host': self.getESHost(), 'port': self.getESPort()}],
                                auth_type=CMRESHandler.AuthType.NO_AUTH,
                                use_ssl=False,
+        handler = CMRESHandler(hosts=[{'host': self.getESHost(),
+                                       'port': self.getESPort(),
+                                       'scheme': self.get_ES_scheme()}],
                                buffer_size=2,
                                flush_frequency_in_sec=1000,
                                es_index_name="pythontest",
@@ -61,9 +70,9 @@ def test_buffered_log_insertion_flushed_when_buffer_full(self):
 
     def test_es_log_extra_argument_insertion(self):
         self.log.info("About to test elasticsearch insertion")
-        handler = CMRESHandler(hosts=[{'host': self.getESHost(), 'port': self.getESPort()}],
                                auth_type=CMRESHandler.AuthType.NO_AUTH,
-                               use_ssl=False,
+        handler = CMRESHandler(hosts=[{'host': self.getESHost(),
+                                       'port': self.getESPort(),
                                es_index_name="pythontest",
                                es_additional_fields={'App': 'Test', 'Environment': 'Dev'},
                                raise_on_indexing_exceptions=True)
@@ -84,9 +93,10 @@ def test_es_log_extra_argument_insertion(self):
         self.assertEqual(0, len(handler._buffer))
 
     def test_buffered_log_insertion_after_interval_expired(self):
-        handler = CMRESHandler(hosts=[{'host': self.getESHost(), 'port': self.getESPort()}],
                                auth_type=CMRESHandler.AuthType.NO_AUTH,
-                               use_ssl=False,
+        handler = CMRESHandler(hosts=[{'host': self.getESHost(),
+                                       'port': self.getESPort(),
+                                       'scheme': self.get_ES_scheme()}],
                                flush_frequency_in_sec=0.1,
                                es_index_name="pythontest",
                                es_additional_fields={'App': 'Test', 'Environment': 'Dev'},
@@ -108,9 +118,10 @@ def test_buffered_log_insertion_after_interval_expired(self):
         self.assertEqual(0, len(handler._buffer))
 
     def test_fast_insertion_of_hundred_logs(self):
-        handler = CMRESHandler(hosts=[{'host': self.getESHost(), 'port': self.getESPort()}],
                                auth_type=CMRESHandler.AuthType.NO_AUTH,
-                               use_ssl=False,
+        handler = CMRESHandler(hosts=[{'host': self.getESHost(),
+                                       'port': self.getESPort(),
+                                       'scheme': self.get_ES_scheme()}],
                                buffer_size=500,
                                flush_frequency_in_sec=0.5,
                                es_index_name="pythontest",
@@ -125,43 +136,47 @@ def test_fast_insertion_of_hundred_logs(self):
 
     def test_index_name_frequency_functions(self):
         index_name = "pythontest"
-        handler = CMRESHandler(hosts=[{'host': self.getESHost(), 'port': self.getESPort()}],
                                auth_type=CMRESHandler.AuthType.NO_AUTH,
+        handler = CMRESHandler(hosts=[{'host': self.getESHost(),
+                                       'port': self.getESPort(),
+                                       'scheme': self.get_ES_scheme()}],
                                es_index_name=index_name,
-                               use_ssl=False,
                                index_name_frequency=CMRESHandler.IndexNameFrequency.DAILY,
                                raise_on_indexing_exceptions=True)
         self.assertEqual(
             handler._index_name_func.__func__(index_name),
             CMRESHandler._get_daily_index_name(index_name)
         )
 
-        handler = CMRESHandler(hosts=[{'host': self.getESHost(), 'port': self.getESPort()}],
                                auth_type=CMRESHandler.AuthType.NO_AUTH,
+        handler = CMRESHandler(hosts=[{'host': self.getESHost(),
+                                       'port': self.getESPort(),
+                                       'scheme': self.get_ES_scheme()}],
                                es_index_name=index_name,
-                               use_ssl=False,
                                index_name_frequency=CMRESHandler.IndexNameFrequency.WEEKLY,
                                raise_on_indexing_exceptions=True)
         self.assertEqual(
             handler._index_name_func.__func__(index_name),
             CMRESHandler._get_weekly_index_name(index_name)
         )
 
-        handler = CMRESHandler(hosts=[{'host': self.getESHost(), 'port': self.getESPort()}],
                                auth_type=CMRESHandler.AuthType.NO_AUTH,
+        handler = CMRESHandler(hosts=[{'host': self.getESHost(),
+                                       'port': self.getESPort(),
+                                       'scheme': self.get_ES_scheme()}],
                                es_index_name=index_name,
-                               use_ssl=False,
                                index_name_frequency=CMRESHandler.IndexNameFrequency.MONTHLY,
                                raise_on_indexing_exceptions=True)
         self.assertEqual(
             handler._index_name_func.__func__(index_name),
             CMRESHandler._get_monthly_index_name(index_name)
         )
 
-        handler = CMRESHandler(hosts=[{'host': self.getESHost(), 'port': self.getESPort()}],
                                auth_type=CMRESHandler.AuthType.NO_AUTH,
+        handler = CMRESHandler(hosts=[{'host': self.getESHost(),
+                                       'port': self.getESPort(),
+                                       'scheme': self.get_ES_scheme()}],
                                es_index_name=index_name,
-                               use_ssl=False,
                                index_name_frequency=CMRESHandler.IndexNameFrequency.YEARLY,
                                raise_on_indexing_exceptions=True)
         self.assertEqual(