Refactor ver1.0.0 (#10) (#11)

* Added variable for selecting index_frequency postfix
* Supports daily, weekly, monthly and yearly postfixes to the indexname
* Changed a map section for an interator when creating actions to reduce
* chances of extra memory used (trade-off with speed)
* Added sonar support

This closes #7, closes #9 . Beta version ready for release

Author: cmanaha

.gitignore

@@ -25,6 +25,12 @@ var/
 .installed.cfg
 *.egg
 
+pylint.txt
+coverage.xml
+pylint_*
+.pylint.d/
+.sonar/
+
 # PyInstaller
 #  Usually these files are written by a python script from a template
 #  before PyInstaller builds the exe, so as to inject date/other infos into it.

.travis.yml

@@ -1,10 +1,11 @@
 language: python
 
-env:
- - TOXENV=py27
+python:
+  - "2.7"
+  - "3.6"
 
 install:
- - pip install tox
+ - pip install tox-travis
  - pip install codecov
  - mkdir /tmp/elasticsearch
  - wget -O - http://s3-eu-west-1.amazonaws.com/build.eu-west-1.elastic.co/origin/2.0/nightly/JDK7/elasticsearch-latest-SNAPSHOT.tar.gz | tar xz --directory=/tmp/elasticsearch --strip-components=1

README.rst

@@ -23,19 +23,30 @@ Install using pip::
 
     pip install CMRESHandler
 
-Requirements
-============
+Requirements Python 2
+=====================
 This library requires the following dependencies
- - requests
- - requests-kerberos
  - elasticsearch
+ - requests
  - enum
 
+
+Requirements Python 3
+=====================
+This library requires the following dependencies
+ - elasticsearch
+ - requests
+
+Additional requirements for Kerberos support
+============================================
+Additionally, the package support optionally kerberos authentication by adding the following dependecy
+ - requests-kerberos
+
 Using the handler in  your program
 ==================================
 To initialise and create the handler, just add the handler to your logger as follow ::
 
-    import CMRESHandler
+    from cmrlogging.handlers import CMRESHandler
     handler = CMRESHandler(hosts=[{'host': 'localhost', 'port': 9200}],
                                auth_type=CMRESHandler.AuthType.NO_AUTH,
                                es_index_name="my_python_index")
@@ -45,7 +56,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 ::
 
-    import CMRESHandler
+    from cmrlogging.handlers import CMRESHandler
     handler = CMRESHandler(hosts=[{'host': 'localhost', 'port': 9200}],
                                auth_type=CMRESHandler.AuthType.NO_AUTH,
                                es_index_name="my_python_index",
@@ -89,6 +100,10 @@ The constructors takes the following parameters:
  - flush_frequency_in_sec: A float representing how often and when the buffer will be flushed
  - es_index_name: A string with the prefix of the elasticsearch index that will be created. Note a date with
    YYYY.MM.dd, ``python_logger`` used by default
+ - index_name_frequency: The frequency to use as part of the index naming. Currently supports
+   CMRESHandler.IndexNameFrequency.DAILY, CMRESHandler.IndexNameFrequency.WEEKLY,
+   CMRESHandler.IndexNameFrequency.MONTHLY, CMRESHandler.IndexNameFrequency.YEARLY by default the daily rotation
+   is used
  - es_doc_type: A string with the name of the document type that will be used ``python_log`` used by default
  - es_additional_fields: A dictionary with all the additional fields that you would like to add to the logs
 
@@ -98,7 +113,7 @@ It is also very easy to integrate the handler to `Django <https://www.djangoproj
 better, at DEBUG level django logs information such as how long it takes for DB connections to return so
 they can be plotted on Kibana, or the SQL statements that Django executed. ::
 
-    from cmreshandler.cmreshandler import CMRESHandler
+    from cmreslogging.handlers import CMRESHandler
     LOGGING = {
         'version': 1,
         'disable_existing_loggers': False,
@@ -112,7 +127,7 @@ they can be plotted on Kibana, or the SQL statements that Django executed. ::
             },
             'elasticsearch': {
                 'level': 'DEBUG',
-                'class': 'cmreshandler.cmreshandler.CMRESHandler',
+                'class': 'cmrlogging.handlers.CMRESHandler',
                 'hosts': [{'host': 'localhost', 'port': 9200}],
                 'es_index_name': 'my_python_app',
                 'es_additional_fields': {'App': 'Test', 'Environment': 'Dev'},

cmreshandler/__init__.py renamed to cmreslogging/__init__.py

@@ -2,13 +2,18 @@
 """
 
 import logging
-from enum import Enum
-from elasticsearch import helpers as eshelpers
-from elasticsearch import Elasticsearch, RequestsHttpConnection
-from requests_kerberos import HTTPKerberosAuth, DISABLED
 import datetime
 import socket
+import copy
 from threading import Timer
+from enum import Enum
+from elasticsearch import helpers as eshelpers
+from elasticsearch import Elasticsearch, RequestsHttpConnection
+try:
+    from requests_kerberos import HTTPKerberosAuth, DISABLED
+    CMR_KERBEROS_SUPPORTED = True
+except ImportError:
+    CMR_KERBEROS_SUPPORTED = False
 
 
 class CMRESHandler(logging.Handler):
@@ -30,38 +35,94 @@ class AuthType(Enum):
         BASIC_AUTH = 1
         KERBEROS_AUTH = 2
 
+    class IndexNameFrequency(Enum):
+        """ Index type supported
+        the handler supports
+        - Daily indices
+        - Weekly indices
+        - Monthly indices
+        - Year indices
+        """
+        DAILY = 0
+        WEEKLY = 1
+        MONTHLY = 2
+        YEARLY = 3
+
     # Defaults for the class
-    __DEFAULT_HOST = [{'host': 'localhost', 'port': 9200}]
+    __DEFAULT_ELASTICSEARCH_HOST = [{'host': 'localhost', 'port': 9200}]
     __DEFAULT_AUTH_USER = ''
     __DEFAULT_AUTH_PASSWD = ''
     __DEFAULT_USE_SSL = False
     __DEFAULT_VERIFY_SSL = True
     __DEFAULT_AUTH_TYPE = AuthType.NO_AUTH
+    __DEFAULT_INDEX_FREQUENCY = IndexNameFrequency.DAILY
     __DEFAULT_BUFFER_SIZE = 1000
-    __DEFAULT_FLUSH_FREQUENCY_IN_SEC = 1
+    __DEFAULT_FLUSH_FREQ_INSEC = 1
     __DEFAULT_ADDITIONAL_FIELDS = {}
     __DEFAULT_ES_INDEX_NAME = 'python_logger'
     __DEFAULT_ES_DOC_TYPE = 'python_log'
-    __DEFAULT_RAISE_ON_INDEXING_EXCEPTIONS = False
+    __DEFAULT_RAISE_ON_EXCEPTION = False
     __DEFAULT_TIMESTAMP_FIELD_NAME = "timestamp"
 
     __LOGGING_FILTER_FIELDS = ['msecs',
                                'relativeCreated',
                                'levelno',
                                'created']
 
+    @staticmethod
+    def _get_daily_index_name(es_index_name):
+        """ Returns elasticearch index name
+        :param: index_name the prefix to be used in the index
+        :return: A srting containing the elasticsearch indexname used which should include the date.
+        """
+        return "{0!s}-{1!s}".format(es_index_name, datetime.datetime.now().strftime('%Y.%m.%d'))
+
+    @staticmethod
+    def _get_weekly_index_name(es_index_name):
+        """ Return elasticsearch index name
+        :param: index_name the prefix to be used in the index
+        :return: A srting containing the elasticsearch indexname used which should include the date and specific week
+        """
+        current_date = datetime.datetime.now()
+        start_of_the_week = current_date - datetime.timedelta(days=current_date.weekday())
+        return "{0!s}-{1!s}".format(es_index_name, start_of_the_week.strftime('%Y.%m.%d'))
+
+    @staticmethod
+    def _get_monthly_index_name(es_index_name):
+        """ Return elasticsearch index name
+        :param: index_name the prefix to be used in the index
+        :return: A srting containing the elasticsearch indexname used which should include the date and specific moth
+        """
+        return "{0!s}-{1!s}".format(es_index_name, datetime.datetime.now().strftime('%Y.%m'))
+
+    @staticmethod
+    def _get_yearly_index_name(es_index_name):
+        """ Return elasticsearch index name
+        :param: index_name the prefix to be used in the index
+        :return: A srting containing the elasticsearch indexname used which should include the date and specific year
+        """
+        return "{0!s}-{1!s}".format(es_index_name, datetime.datetime.now().strftime('%Y'))
+
+    _INDEX_FREQUENCY_FUNCION_DICT = {
+        IndexNameFrequency.DAILY: _get_daily_index_name,
+        IndexNameFrequency.WEEKLY: _get_weekly_index_name,
+        IndexNameFrequency.MONTHLY: _get_monthly_index_name,
+        IndexNameFrequency.YEARLY: _get_yearly_index_name
+    }
+
     def __init__(self,
-                 hosts=__DEFAULT_HOST,
+                 hosts=__DEFAULT_ELASTICSEARCH_HOST,
                  auth_details=(__DEFAULT_AUTH_USER, __DEFAULT_AUTH_PASSWD),
                  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_FREQUENCY_IN_SEC,
+                 flush_frequency_in_sec=__DEFAULT_FLUSH_FREQ_INSEC,
                  es_index_name=__DEFAULT_ES_INDEX_NAME,
+                 index_name_frequency=__DEFAULT_INDEX_FREQUENCY,
                  es_doc_type=__DEFAULT_ES_DOC_TYPE,
                  es_additional_fields=__DEFAULT_ADDITIONAL_FIELDS,
-                 raise_on_indexing_exceptions=__DEFAULT_RAISE_ON_INDEXING_EXCEPTIONS,
+                 raise_on_indexing_exceptions=__DEFAULT_RAISE_ON_EXCEPTION,
                  default_timestamp_field_name=__DEFAULT_TIMESTAMP_FIELD_NAME):
         """ Handler constructor
 
@@ -80,6 +141,10 @@ def __init__(self,
                     if the buffer_size has not been reached yet
         :param es_index_name: A string with the prefix of the elasticsearch index that will be created. Note a
                     date with YYYY.MM.dd, ```python_logger``` used by default
+        :param index_name_frequency: Defines what the date used in the postfix of the name would be. available values
+                    are selected from the IndexNameFrequency class (IndexNameFrequency.DAILY,
+                    IndexNameFrequency.WEEKLY, IndexNameFrequency.MONTHLY, IndexNameFrequency.YEARLY). By default
+                    it uses daily indices.
         :param es_doc_type: A string with the name of the document type that will be used ```python_log``` used
                     by default
         :param es_additional_fields: A dictionary with all the additional fields that you would like to add
@@ -90,25 +155,29 @@ def __init__(self,
         """
         logging.Handler.__init__(self)
 
-        self.hosts = hosts
+        self.hosts = copy.deepcopy(hosts)
         self.auth_details = auth_details
         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
         self.es_index_name = es_index_name
+        self.index_name_frequency = index_name_frequency
         self.es_doc_type = es_doc_type
-        self.es_additional_fields = es_additional_fields.copy()
+        self.es_additional_fields = copy.deepcopy(es_additional_fields)
         self.es_additional_fields.update({'host': socket.gethostname(),
                                           'host_ip': socket.gethostbyname(socket.gethostname())})
         self.raise_on_indexing_exceptions = raise_on_indexing_exceptions
         self.default_timestamp_field_name = default_timestamp_field_name
 
+        self._client = None
         self._buffer = []
         self._timer = None
         self.__schedule_flush()
 
+        self._index_name_func = CMRESHandler._INDEX_FREQUENCY_FUNCION_DICT[self.index_name_frequency]
+
     def __schedule_flush(self):
         if self._timer is None:
             self._timer = Timer(self.flush_frequency_in_sec, self.flush)
@@ -117,22 +186,34 @@ def __schedule_flush(self):
 
     def __get_es_client(self):
         if self.auth_type == CMRESHandler.AuthType.NO_AUTH:
-            return Elasticsearch(hosts=self.hosts,
-                                 use_ssl=self.use_ssl,
-                                 verify_certs=self.verify_certs,
-                                 connection_class=RequestsHttpConnection)
-        elif self.auth_type == CMRESHandler.AuthType.BASIC_AUTH:
-            return Elasticsearch(hosts=self.hosts,
-                                 http_auth=self.auth_details,
-                                 use_ssl=self.use_ssl,
-                                 verify_certs=self.verify_certs,
-                                 connection_class=RequestsHttpConnection)
-        elif self.auth_type == CMRESHandler.AuthType.KERBEROS_AUTH:
-            return Elasticsearch(hosts=self.hosts,
-                                 use_ssl=self.use_ssl,
-                                 verify_certs=self.verify_certs,
-                                 connection_class=RequestsHttpConnection,
-                                 http_auth=HTTPKerberosAuth(mutual_authentication=DISABLED))
+            if self._client is None:
+                self._client = Elasticsearch(hosts=self.hosts,
+                                             use_ssl=self.use_ssl,
+                                             verify_certs=self.verify_certs,
+                                             connection_class=RequestsHttpConnection)
+            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)
+            return self._client
+
+        if self.auth_type == CMRESHandler.AuthType.KERBEROS_AUTH:
+            if CMR_KERBEROS_SUPPORTED:
+                # 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))
+            else:
+                raise EnvironmentError("Kerberos module not available. Please install \"requests-kerberos\"")
+
+        raise ValueError("Authentication method not supported")
 
     def test_es_source(self):
         """ Returns True if the handler can ping the Elasticsearch servers
@@ -144,21 +225,15 @@ def test_es_source(self):
         """
         return self.__get_es_client().ping()
 
-    def __get_es_index_name(self):
-        """ Returns elasticearch index name
-        :return: A srting containing the elasticsearch indexname used which should include the date.
-        """
-        return "{0!s}-{1!s}".format(self.es_index_name, datetime.datetime.now().strftime('%Y.%m.%d'))
-
     @staticmethod
     def __get_es_datetime_str(timestamp):
         """ Returns elasticsearch utc formatted time for an epoch timestamp
 
         :param timestamp: epoch, including milliseconds
         :return: A string valid for elasticsearch time record
         """
-        t = datetime.datetime.utcfromtimestamp(timestamp)
-        return "{0!s}.{1:03d}Z".format(t.strftime('%Y-%m-%dT%H:%M:%S'), int(t.microsecond / 1000))
+        current_date = datetime.datetime.utcfromtimestamp(timestamp)
+        return "{0!s}.{1:03d}Z".format(current_date.strftime('%Y-%m-%dT%H:%M:%S'), int(current_date.microsecond / 1000))
 
     def flush(self):
         """ Flushes the buffer into ES
@@ -168,19 +243,25 @@ def flush(self):
             self._timer.cancel()
         self._timer = None
 
-        # FIXME: This should probably go on a different thread to speed up the execution
         if len(self._buffer) >= 0:
             try:
-                actions = map(lambda x: {'_index': self.__get_es_index_name(),
-                                         '_type': self.es_doc_type,
-                                         '_source': x},
-                              self._buffer)
-                eshelpers.bulk(client=self.__get_es_client(),
-                               actions=actions,
-                               stats_only=True)
-            except Exception as e:
+                actions = (
+                    {
+                        '_index': self._index_name_func.__func__(self.es_index_name),
+                        '_type': self.es_doc_type,
+                        '_source': log_record
+                    }
+                    for log_record in self._buffer
+                )
+
+                eshelpers.bulk(
+                    client=self.__get_es_client(),
+                    actions=actions,
+                    stats_only=True
+                )
+            except Exception as exception:
                 if self.raise_on_indexing_exceptions:
-                    raise e
+                    raise exception
             self._buffer = []
 
         self.__schedule_flush()
@@ -203,9 +284,9 @@ def emit(self, record):
         :return: None
         """
         rec = self.es_additional_fields.copy()
-        for k, v in record.__dict__.items():
-            if k not in CMRESHandler.__LOGGING_FILTER_FIELDS:
-                rec[k] = "" if v is None else v
+        for key, value in record.__dict__.items():
+            if key not in CMRESHandler.__LOGGING_FILTER_FIELDS:
+                rec[key] = "" if value is None else value
         rec[self.default_timestamp_field_name] = self.__get_es_datetime_str(record.created)
 
         self._buffer.append(rec)

cmreshandler/cmreshandler.py renamed to cmreslogging/handlers.py

@@ -0,0 +1,3 @@
+elasticsearch==5.1.0
+requests==2.12.4
+enum==0.4.6

requirements/requirements_py27.txt

@@ -0,0 +1,2 @@
+elasticsearch==5.1.0
+requests==2.12.4