| """Service class, CanaryDocumentGenerator class. |
| |
| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| """ |
| |
| import logging |
| import os |
| import time |
| |
| from consumer import Consumer |
| from database import Database |
| from datetime import datetime |
| from datetimeutils import secondsSince |
| from requests.exceptions import ConnectionError, ReadTimeout |
| from threading import Thread |
| |
| # How often to produce canary documents |
| canaryInterval = 60 # seconds |
| |
| # How long the changes feed should poll before timing out |
| changesFeedTimeout = 30 # seconds |
| |
| |
| class Service (Thread): |
| def __init__(self, consumers): |
| Thread.__init__(self) |
| self.daemon = True |
| |
| self.database = None |
| self.lastSequence = None |
| self.canaryGenerator = CanaryDocumentGenerator() |
| |
| self.consumers = consumers |
| self.workerId = os.getenv("WORKER", "worker0") |
| |
| def run(self): |
| self.canaryGenerator.start() |
| self.lastCanaryTime = datetime.now() |
| |
| while True: |
| try: |
| if self.database is not None: |
| logging.info("Shutting down existing DB client") |
| self.database.destroy() |
| |
| logging.info("Starting changes feed") |
| self.database = Database(timeout=changesFeedTimeout) |
| self.changes = self.database.changesFeed(timeout=changesFeedTimeout, since=self.lastSequence) |
| |
| for change in self.changes: |
| # change could be None because the changes feed will timeout |
| # if it hasn't detected any changes. This timeout allows us to |
| # check whether or not the feed is capable of detecting canary |
| # documents |
| if change != None: |
| self.__handleDocChange(change) |
| |
| # Record the sequence in case the changes feed needs to be |
| # restarted. This way the new feed can pick up right where |
| # the old one left off. |
| self.lastSequence = change['seq'] |
| except Exception as e: |
| logging.error('[canary] Exception caught from changes feed. Restarting changes feed...') |
| logging.error(e) |
| self.stopChangesFeed() |
| |
| logging.debug("[changes] I made it out of the changes loop!") |
| |
| def __handleDocChange(self, change): |
| retry = True |
| retryCount = 0 |
| maxRetries = 5 |
| |
| while retry: |
| try: |
| if "deleted" in change and change["deleted"] == True: |
| logging.info('[changes] Found a delete') |
| consumer = self.consumers.getConsumerForTrigger(change['id']) |
| if consumer != None: |
| if consumer.desiredState() == Consumer.State.Disabled: |
| # just remove it from memory |
| logging.info('[{}] Removing disabled trigger'.format(consumer.trigger)) |
| self.consumers.removeConsumerForTrigger(consumer.trigger) |
| else: |
| logging.info('[{}] Shutting down running trigger'.format(consumer.trigger)) |
| consumer.shutdown() |
| # since we can't use a filter function for the feed (then |
| # you don't get deletes) we need to manually verify this |
| # is a valid trigger doc that has changed |
| elif 'triggerURL' in change['doc']: |
| logging.info('[changes] Found a change in a trigger document') |
| document = change['doc'] |
| triggerIsAssignedToMe = self.__isTriggerDocAssignedToMe(document) |
| |
| if not self.consumers.hasConsumerForTrigger(change["id"]): |
| if triggerIsAssignedToMe: |
| logging.info('[{}] Found a new trigger to create'.format(change["id"])) |
| self.createAndRunConsumer(document) |
| else: |
| logging.info("[{}] Found a new trigger, but is assigned to another worker: {}".format(change["id"], document["worker"])) |
| else: |
| existingConsumer = self.consumers.getConsumerForTrigger(change["id"]) |
| |
| if existingConsumer.desiredState() == Consumer.State.Running and not self.__isTriggerDocActive(document): |
| # running trigger should become disabled |
| # this should be done regardless of which worker the document claims to be assigned to |
| logging.info('[{}] Existing running trigger should become disabled'.format(change["id"])) |
| existingConsumer.disable() |
| elif triggerIsAssignedToMe: |
| logging.info('[{}] Found a change to an existing trigger'.format(change["id"])) |
| |
| if existingConsumer.desiredState() == Consumer.State.Dead and self.__isTriggerDocActive(document): |
| # if a delete occurs followed quickly by a create the consumer might get stuck in a dead state, |
| # so we need to forcefully delete the process before recreating it. |
| logging.info('[{}] A create event occurred for a trigger that is shutting down'.format(change["id"])) |
| |
| if existingConsumer.process.is_alive(): |
| logging.info('[{}] Joining dead process.'.format(existingConsumer.trigger)) |
| existingConsumer.process.join(1) |
| else: |
| logging.info('[{}] Process is already dead.'.format(existingConsumer.trigger)) |
| |
| self.consumers.removeConsumerForTrigger(existingConsumer.trigger) |
| self.createAndRunConsumer(document) |
| elif existingConsumer.desiredState() == Consumer.State.Disabled and self.__isTriggerDocActive(document): |
| # disabled trigger has become active |
| logging.info('[{}] Existing disabled trigger should become active'.format(change["id"])) |
| self.createAndRunConsumer(document) |
| else: |
| # trigger has become reassigned to a different worker |
| logging.info("[{}] Shutting down trigger as it has been re-assigned to {}".format(change["id"], document["worker"])) |
| existingConsumer.shutdown() |
| elif 'canary-timestamp' in change['doc']: |
| # found a canary - update lastCanaryTime |
| logging.info('[canary] I found a canary. The last one was {} seconds ago.'.format(secondsSince(self.lastCanaryTime))) |
| self.lastCanaryTime = datetime.now() |
| else: |
| logging.debug('[changes] Found a change for a non-trigger document') |
| |
| retry = False |
| except Exception as e: |
| logging.error('[{}] Exception caught while handling change.'.format(change["id"])) |
| logging.error(e) |
| |
| if retry: |
| retryCount += 1 |
| |
| if retryCount >= maxRetries: |
| logging.warn('[{}] Maximum number of retries exceeded for failed change.'.format(change["id"])) |
| retry = False |
| |
| def __isTriggerDocAssignedToMe(self, doc): |
| if "worker" in doc: |
| return doc["worker"] == self.workerId |
| else: |
| return self.workerId == "worker0" |
| |
| def stopChangesFeed(self): |
| if self.changes != None: |
| self.changes.stop() |
| self.changes = None |
| |
| def createAndRunConsumer(self, doc): |
| triggerFQN = doc['_id'] |
| |
| # Create a representation for this trigger, even if it is disabled |
| # This allows it to appear in /health as well as allow it to be deleted |
| # Creating this object is lightweight and does not initialize any connections |
| consumer = Consumer(triggerFQN, doc) |
| self.consumers.addConsumerForTrigger(triggerFQN, consumer) |
| |
| if self.__isTriggerDocActive(doc): |
| logging.info('[{}] Trigger was determined to be active, starting...'.format(triggerFQN)) |
| consumer.start() |
| else: |
| logging.info('[{}] Trigger was determined to be disabled, not starting...'.format(triggerFQN)) |
| |
| def __isTriggerDocActive(self, doc): |
| return ('status' not in doc or doc['status']['active'] == True) |
| |
| |
| class CanaryDocumentGenerator (Thread): |
| def __init__(self): |
| Thread.__init__(self) |
| self.daemon = True |
| self.database = Database() |
| |
| def run(self): |
| while True: |
| # create a new canary document every so often |
| self.database.createCanary() |
| time.sleep(canaryInterval) |
| |
| logging.error('[canary generator] Exited the main loop!') |
