Merge branch 'downstream' into devel

awx/main/dispatch/worker/base.py

@@ -8,6 +8,7 @@ import sys
 import redis
 import json
 import psycopg2
+import time
 from uuid import UUID
 from queue import Empty as QueueEmpty
 
@@ -116,18 +117,23 @@ class AWXConsumerRedis(AWXConsumerBase):
         super(AWXConsumerRedis, self).run(*args, **kwargs)
         self.worker.on_start()
 
-        queue = redis.Redis.from_url(settings.BROKER_URL)
+        time_to_sleep = 1
         while True:
-            try:
-                res = queue.blpop(self.queues)
-                res = json.loads(res[1])
-                self.process_task(res)
-            except redis.exceptions.RedisError:
-                logger.exception("encountered an error communicating with redis")
-            except (json.JSONDecodeError, KeyError):
-                logger.exception("failed to decode JSON message from redis")
-            if self.should_stop:
-                return
+            queue = redis.Redis.from_url(settings.BROKER_URL)
+            while True:
+                try:
+                    res = queue.blpop(self.queues)
+                    time_to_sleep = 1
+                    res = json.loads(res[1])
+                    self.process_task(res)
+                except redis.exceptions.RedisError:
+                    time_to_sleep = min(time_to_sleep * 2, 30)
+                    logger.exception(f"encountered an error communicating with redis. Reconnect attempt in {time_to_sleep} seconds")
+                    time.sleep(time_to_sleep)
+                except (json.JSONDecodeError, KeyError):
+                    logger.exception("failed to decode JSON message from redis")
+                if self.should_stop:
+                    return
 
 
 class AWXConsumerPG(AWXConsumerBase):

awx/main/tasks.py

@@ -1232,10 +1232,12 @@ class BaseTask(object):
             # this is a _little_ expensive to filter
             # with regex, but project updates don't have many events,
             # so it *should* have a negligible performance impact
+            task = event_data.get('event_data', {}).get('task_action')
             try:
-                event_data_json = json.dumps(event_data)
-                event_data_json = UriCleaner.remove_sensitive(event_data_json)
-                event_data = json.loads(event_data_json)
+                if task in ('git', 'hg', 'svn'):
+                    event_data_json = json.dumps(event_data)
+                    event_data_json = UriCleaner.remove_sensitive(event_data_json)
+                    event_data = json.loads(event_data_json)
             except json.JSONDecodeError:
                 pass