python/google/appengine/tools/devappserver2/health_check_service.py

   1 #!/usr/bin/env python
   2 #
   3 # Copyright 2007 Google Inc.
   4 #
   5 # Licensed under the Apache License, Version 2.0 (the "License");
   6 # you may not use this file except in compliance with the License.
   7 # You may obtain a copy of the License at
   8 #
   9 #     http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 # Unless required by applicable law or agreed to in writing, software
  12 # distributed under the License is distributed on an "AS IS" BASIS,
  13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16 #
  17 """A health checking implementation for the SDK.
  18
  19 <scrub>
  20 This code attempts to match the production implementation as closely as
  21 possible in apphosting/runtime/vm/vm_health_check.cc.
  22
  23 One instance of HealthChecker should be created per instance.Instance that has
  24 health checking enabled. The HealthChecker instance needs to be started, but
  25 will stop itself automatically.
  26 </scrub>
  27 """
  28
  29
  30 import logging
  31 import threading
  32 import time
  33
  34 from google.appengine.api import request_info
  35 from google.appengine.tools.devappserver2 import start_response_utils
  36
  37
  38 class _HealthCheckState(object):
  39   """A class to track the state of a health checked instance."""
  40
  41   def __init__(self):
  42     """Initializes a _HealthCheckState object."""
  43     self.consecutive_healthy_responses = 0
  44     self.consecutive_unhealthy_responses = 0
  45     self.is_last_successful = False
  46
  47   def update(self, healthy):
  48     """Updates the state.
  49
  50     Args:
  51       healthy: Bool indicating whether the last attempt was healthy.
  52     """
  53     self.is_last_successful = healthy
  54     if healthy:
  55       self.consecutive_healthy_responses += 1
  56       self.consecutive_unhealthy_responses = 0
  57     else:
  58       self.consecutive_healthy_responses = 0
  59       self.consecutive_unhealthy_responses += 1
  60
  61   def __repr__(self):
  62     """Outputs the state in a readable way for logging."""
  63     tmpl = '{number} consecutive {state} responses.'
  64     if self.consecutive_healthy_responses:
  65       number = self.consecutive_healthy_responses
  66       state = 'HEALTHY'
  67     else:
  68       number = self.consecutive_unhealthy_responses
  69       state = 'UNHEALTHY'
  70     return tmpl.format(numer=number, state=state)
  71
  72
  73 class HealthChecker(object):
  74   """A class to perform health checks for an instance.
  75
  76   This class uses the settings specified in appinfo.VmHealthCheck and the
  77   callback specified to check the health of the specified instance. When
  78   appropriate, this class changes the state of the specified instance so it is
  79   placed into or taken out of load balancing. This class will also use another
  80   callback to restart the instance, if necessary.
  81   """
  82
  83   def __init__(self, instance, config, send_request, restart):
  84     """Initializes a HealthChecker object.
  85
  86     Args:
  87       instance: An instance.Instance object.
  88       config: An appinfo.VmHealthCheck object.
  89       send_request: A function to call that makes the health check request.
  90       restart: A function to call that restarts the instance.
  91     """
  92     self._instance = instance
  93     self._config = config
  94     self._send_request = send_request
  95     self._restart = restart
  96
  97   def start(self):
  98     """Starts the health checks."""
  99     logging.info('Health checks starting for instance %s.',
 100                  self._instance.instance_id)
 101     loop = threading.Thread(target=self._loop)
 102     loop.daemon = True
 103     loop.start()
 104
 105   def _should_continue(self):
 106     return self._running and not self._instance.has_quit
 107
 108   def _loop(self):
 109     """Performs health checks and updates state over time."""
 110     state = _HealthCheckState()
 111     self._instance.set_health(False)
 112     self._running = True
 113     while self._should_continue():
 114       logging.debug('Performing health check for instance %s.',
 115                     self._instance.instance_id)
 116       self._do_health_check(state)
 117       logging.debug('Health check state for instance: %s: %r',
 118                     self._instance.instance_id, state)
 119       time.sleep(self._config.check_interval_sec)
 120
 121   def _do_health_check(self, state):
 122     health = self._get_health_check_response(state.is_last_successful)
 123     state.update(health)
 124     self._maybe_update_instance(state)
 125
 126   def _maybe_update_instance(self, state):
 127     """Performs any required actions on the instance based on the state.
 128
 129     Args:
 130       state: A _HealthCheckState object.
 131     """
 132     if (state.consecutive_unhealthy_responses >=
 133         self._config.unhealthy_threshold):
 134       self._instance.set_health(False)
 135     elif (state.consecutive_healthy_responses >=
 136           self._config.healthy_threshold):
 137       self._instance.set_health(True)
 138
 139     if (state.consecutive_unhealthy_responses >=
 140         self._config.restart_threshold):
 141       self._restart_instance()
 142
 143   def _get_health_check_response(self, is_last_successful):
 144     """Sends the health check request and checks the result.
 145
 146     Args:
 147       is_last_successful: Whether the last request was successful.
 148
 149     Returns:
 150       A bool indicating whether or not the instance is healthy.
 151     """
 152     start_response = start_response_utils.CapturingStartResponse()
 153     try:
 154       response = self._send_request(start_response, is_last_successful)
 155     except request_info.Error, e:
 156       logging.warning('Health check failure for instance %s. Exception: %s',
 157                       self._instance.instance_id, e)
 158       return False
 159     logging.debug('Health check response %s and status %s for instance %s.',
 160                   response, start_response.status, self._instance.instance_id)
 161
 162     return response == ['ok'] and start_response.status == '200 OK'
 163
 164   def _restart_instance(self):
 165     """Restarts the running instance, and stops the current health checker."""
 166     logging.warning('Restarting instance %s because of failed health checks.',
 167                     self._instance.instance_id)
 168     self._running = False
 169     self._restart()