4.4.1 - workaround for issue #31

Don't crash if Elasticsearch returns an unexpected result

CHANGELOG.md

@@ -1,3 +1,8 @@
+4.4.1
+-----
+
+- Don't crash if Elasticsearch returns an unexpected result (workaround for issue #31)
+
 4.4.0
 -----
 

parsedmarc/cli.py

@@ -42,10 +42,9 @@ def _main():
                             report, index=es_aggregate_index)
                 except elastic.AlreadySaved as warning:
                     logger.warning(warning.__str__())
-                except ElasticsearchException as error_:
+                except elastic.ElasticsearchError as error_:
                     logger.error("Elasticsearch Error: {0}".format(
                         error_.__str__()))
-                    exit(1)
                 try:
                     if args.kafka_hosts:
                         kafka_client.save_aggregate_reports_to_kafka(
@@ -69,7 +68,7 @@ def _main():
                             report, index=es_forensic_index)
                 except elastic.AlreadySaved as warning:
                     logger.warning(warning.__str__())
-                except ElasticsearchException as error_:
+                except elastic.ElasticsearchError as error_:
                     logger.error("Elasticsearch Error: {0}".format(
                         error_.__str__()))
                 try:
@@ -243,7 +242,7 @@ def _main():
             if args.elasticsearch_host:
                 elastic.set_hosts(args.elasticsearch_host)
                 elastic.create_indexes([es_aggregate_index, es_forensic_index])
-        except ElasticsearchException as error:
+        except elastic.ElasticsearchError as error:
             logger.error("Elasticsearch Error: {0}".format(error.__str__()))
             exit(1)
 

parsedmarc/elastic.py

@@ -12,6 +12,10 @@ from parsedmarc.utils import human_timestamp_to_datetime
 logger = logging.getLogger("parsedmarc")
 
 
+class ElasticsearchError(Exception):
+    """Raised when an Elasticsearch error occurs"""
+
+
 class _PolicyOverride(InnerDoc):
     type = Text()
     comment = Text()
@@ -187,11 +191,15 @@ def create_indexes(names=None, settings=None):
         names = ["dmarc_aggregate", "dmarc_forensic"]
     for name in names:
         index = Index(name)
-        if not index.exists():
-            logger.debug("Creating Elasticsearch index: {0}".format(name))
-            if settings:
-                index.put_settings(settings)
-            index.create()
+        try:
+            if not index.exists():
+                logger.debug("Creating Elasticsearch index: {0}".format(name))
+                if settings:
+                    index.put_settings(settings)
+                index.create()
+        except Exception as e:
+            raise ElasticsearchError(
+                "Elasticsearch error: {0}".format(e.__str__()))
 
 
 def save_aggregate_report_to_elasticsearch(aggregate_report,
@@ -289,7 +297,11 @@ def save_aggregate_report_to_elasticsearch(aggregate_report,
                                    result=spf_result["result"])
 
         agg_doc.meta.index = index
-        agg_doc.save()
+        try:
+            agg_doc.save()
+        except Exception as e:
+            raise ElasticsearchError(
+                "Elasticsearch error: {0}".format(e.__str__()))
 
 
 def save_forensic_report_to_elasticsearch(forensic_report,
@@ -401,4 +413,8 @@ def save_forensic_report_to_elasticsearch(forensic_report,
     )
 
     forensic_doc.meta.index = index
-    forensic_doc.save()
+    try:
+        forensic_doc.save()
+    except Exception as e:
+        raise ElasticsearchError(
+            "Elasticsearch error: {0}".format(e.__str__()))