Disable triggers for invalid auth when using custom auth handler

provider/authHandler.py

@@ -23,6 +23,9 @@
 
 from requests.auth import AuthBase
 
+class AuthHandlerException(Exception):
+    def __init__(self, response):
+        self.response = response
 
 class IAMAuth(AuthBase):
 
@@ -35,18 +38,24 @@ def __call__(self, r):
         r.headers['Authorization'] = 'Bearer {}'.format(self.__getToken())
         return r
 
-
     def __getToken(self):
         if 'expires_in' not in self.tokenInfo or self.__isRefreshTokenExpired():
-            self.tokenInfo = self.__requestToken()
-            return self.tokenInfo['access_token']
+            response = self.__requestToken()
+            if response.ok and 'access_token' in response.json():
+                self.tokenInfo = response.json()
+                return self.tokenInfo['access_token']
+            else:
+                raise AuthHandlerException(response)
         elif self.__isTokenExpired():
-            self.tokenInfo = self.__refreshToken()
-            return self.tokenInfo['access_token']
+            response = self.__refreshToken()
+            if response.ok and 'access_token' in response.json():
+                self.tokenInfo = response.json()
+                return self.tokenInfo['access_token']
+            else:
+                raise AuthHandlerException(response)
         else:
             return self.tokenInfo['access_token']
 
-
     def __requestToken(self):
         headers = {
             'Content-type': 'application/x-www-form-urlencoded',
@@ -86,7 +95,7 @@ def __isTokenExpired(self):
 
     def __isRefreshTokenExpired(self):
         if 'expiration' not in self.tokenInfo:
-            return true
+            return True
 
         sevenDays = 7 * 24 * 3600
         currentTime = int(time.time())
@@ -96,4 +105,4 @@ def __isRefreshTokenExpired(self):
 
     def __sendRequest(self, payload, headers):
         response = requests.post(self.endpoint, data=payload, headers=headers)
-        return response.json()
+        return response

provider/consumer.py

@@ -32,6 +32,7 @@
 from datetimeutils import secondsSince
 from multiprocessing import Process, Manager
 from urlparse import urlparse
+from authHandler import AuthHandlerException
 from authHandler import IAMAuth
 from requests.auth import HTTPBasicAuth
 from datetime import datetime, timedelta
@@ -409,44 +410,19 @@ def __fireTrigger(self, messages):
                         self.consumer.commit(offsets=self.__getOffsetList(messages), async=False)
                         retry = False
                     elif self.__shouldDisable(status_code):
-                        logging.error('[{}] Error talking to OpenWhisk, status code {}'.format(self.trigger, status_code))
-                        response_dump = {
-                            'request': {
-                                'method': response.request.method,
-                                'url': response.request.url,
-                                'path_url': response.request.path_url,
-                                'headers': response.request.headers,
-                                'body': response.request.body
-                            },
-                            'response': {
-                                'status_code': response.status_code,
-                                'ok': response.ok,
-                                'reason': response.reason,
-                                'url': response.url,
-                                'headers': response.headers,
-                                'content': response.content
-                            }
-                        }
-
-                        logging.error('[{}] Dumping the content of the request and response:\n{}'.format(self.trigger, response_dump))
-
-                        # abandon all hope?
-                        self.setDesiredState(Consumer.State.Disabled)
-                        # mark it disabled in the DB
-
-                        # when failing to establish a database connection, mark the consumer as dead to restart the consumer
-                        try:
-                            self.database = Database()
-                            self.database.disableTrigger(self.trigger, status_code)
-                        except Exception as e:
-                            logging.error('[{}] Uncaught exception: {}'.format(self.trigger, e))
-                            self.__recordState(Consumer.State.Dead)
-                        finally:
-                            self.database.destroy()
-
                         retry = False
+                        logging.error('[{}] Error talking to OpenWhisk, status code {}'.format(self.trigger, status_code))
+                        self.__dumpRequestResponse(response)
+                        self.__disableTrigger(status_code)
                 except requests.exceptions.RequestException as e:
                     logging.error('[{}] Error talking to OpenWhisk: {}'.format(self.trigger, e))
+                except AuthHandlerException as e:
+                    logging.error("[{}] Encountered an exception from auth handler, status code {}").format(self.trigger, e.response.status_code)
+                    self.__dumpRequestResponse(e.response)
+
+                    if self.__shouldDisable(e.response.status_code):
+                        retry = False
+                        self.__disableTrigger(e.response.status_code)
 
                 if retry:
                     retry_count += 1
@@ -460,6 +436,40 @@ def __fireTrigger(self, messages):
                         self.consumer.commit(offsets=self.__getOffsetList(messages), async=False)
                         retry = False
 
+    def __disableTrigger(self, status_code):
+        self.setDesiredState(Consumer.State.Disabled)
+
+        # when failing to establish a database connection, mark the consumer as dead to restart the consumer
+        try:
+            self.database = Database()
+            self.database.disableTrigger(self.trigger, status_code)
+        except Exception as e:
+            logging.error('[{}] Uncaught exception: {}'.format(self.trigger, e))
+            self.__recordState(Consumer.State.Dead)
+        finally:
+            self.database.destroy()
+
+    def __dumpRequestResponse(self, response):
+        response_dump = {
+            'request': {
+                'method': response.request.method,
+                'url': response.request.url,
+                'path_url': response.request.path_url,
+                'headers': response.request.headers,
+                'body': response.request.body
+            },
+            'response': {
+                'status_code': response.status_code,
+                'ok': response.ok,
+                'reason': response.reason,
+                'url': response.url,
+                'headers': response.headers,
+                'content': response.content
+            }
+        }
+
+        logging.error('[{}] Dumping the content of the request and response:\n{}'.format(self.trigger, response_dump))
+
     # return the dict that will be sent as the trigger payload
     def __getMessagePayload(self, message):
         return {