From cc5d9265ec7ecd081a39d440b748d962834db81b Mon Sep 17 00:00:00 2001
From: Clark Boylan <clark.boylan@gmail.com>
Date: Mon, 10 Feb 2014 10:20:12 -0800
Subject: [PATCH] Handle log filter exceptions more gracefully.

If there is an exception filtering a log event handle that by removing
the filter and continuing to process the remaining log events for the
assocaited file. This prevents non filter data from being lost when the
filters have an exception.

Change-Id: I65141daf21a873096829c41fdc2c77cbeecde2e3
---
 files/log-gearman-worker.py | 22 +++++++++++++++++-----
 1 file changed, 17 insertions(+), 5 deletions(-)

diff --git a/files/log-gearman-worker.py b/files/log-gearman-worker.py
index 6425099..23ca4dc 100644
--- a/files/log-gearman-worker.py
+++ b/files/log-gearman-worker.py
@@ -52,6 +52,10 @@ def semi_busy_wait(seconds):
             return
 
 
+class FilterException(Exception):
+    pass
+
+
 class CRM114Filter(object):
     def __init__(self, script, path, build_status):
         self.p = None
@@ -77,14 +81,14 @@ class CRM114Filter(object):
                                   [self.p.stdin, self.p.stdout], 20)
         if not r:
             self.p.kill()
-            raise Exception('Timeout reading from CRM114')
+            raise FilterException('Timeout reading from CRM114')
         r = self.p.stdout.readline()
         if not r:
             err = self.p.stderr.read()
             if err:
-                raise Exception(err)
+                raise FilterException(err)
             else:
-                raise Exception('Early EOF from CRM114')
+                raise FilterException('Early EOF from CRM114')
         r = r.strip()
         data['error_pr'] = float(r)
 
@@ -143,6 +147,7 @@ class LogRetriever(threading.Thread):
                 for f in self.filters:
                     logging.debug("Adding filter: %s" % f.name)
                     filters.append(f.create(fields))
+                all_filters = filters
 
                 logging.debug("Pushing " + str(len(log_lines)) + " log lines.")
                 base_event = {}
@@ -151,10 +156,17 @@ class LogRetriever(threading.Thread):
                 for line in log_lines:
                     out_event = base_event.copy()
                     out_event["message"] = line
+                    new_filters = []
                     for f in filters:
-                        f.process(out_event)
+                        try:
+                            f.process(out_event)
+                            new_filters.append(f)
+                        except FilterException:
+                            logging.exception("Exception filtering event: "
+                                              "%s" % line.encode("utf-8"))
+                    filters = new_filters
                     self.logq.put(out_event)
-                for f in filters:
+                for f in all_filters:
                     f.close()
             job.sendWorkComplete()
         except Exception as e: