def _perform_check_if_not_disabled(self):
    if self.snooze_file and os.path.isfile(self.snooze_file):
      log.info("Health check snooze file found at %s. Health checks disabled.", self.snooze_file)
      return True, None

    log.debug("Health checks enabled. Performing health check.")
    return self.checker()

  def is_alive(self):
    """
      Is the process underlying the Thermos task runner alive?
    """
    if not self._popen:
      return False
    if self._dead.is_set():
      return False

    # N.B. You cannot mix this code and any code that relies upon os.wait
    # mechanisms with blanket child process collection.  One example is the
    # Thermos task runner which calls os.wait4 -- without refactoring, you
    # should not mix a Thermos task runner in the same process as this
    # thread.
    try:
      pid, _ = os.waitpid(self._popen.pid, os.WNOHANG)
      if pid == 0:
        return True
      else:
        log.info('Detected runner termination: pid=%s' % pid)
    except OSError as e:
      log.error('is_alive got OSError: %s' % e)
      if e.errno != errno.ECHILD:
        raise

    self._dead.set()
    return False

  def _rollback(self, instances_to_rollback, instance_configs):
    """Performs a rollback operation for the failed instances.

    Arguments:
    instances_to_rollback -- instance ids to rollback.
    instance_configs -- instance configuration to use for rollback.
    """
    if not self._update_config.rollback_on_failure:
      log.info('Rollback on failure is disabled in config. Aborting rollback')
      return

    log.info('Reverting update for %s' % instances_to_rollback)
    instance_operation = self.OperationConfigs(
        from_config=instance_configs.local_config_map,
        to_config=instance_configs.remote_config_map
    )
    instances_to_rollback.sort(reverse=True)
    failed_instances = []
    while instances_to_rollback:
      batch_instances = instances_to_rollback[0 : self._update_config.batch_size]
      instances_to_rollback = list(set(instances_to_rollback) - set(batch_instances))
      instances_to_rollback.sort(reverse=True)
      instances_to_watch = self._update_instances(batch_instances, instance_operation)
      failed_instances += self._watcher.watch(instances_to_watch)

    if failed_instances:
      log.error('Rollback failed for instances: %s' % failed_instances)

def increase_quota(cluster, role, cpu_str, ram_str, disk_str):
  """usage: increase_quota cluster role cpu ram[unit] disk[unit]

  Increases the amount of production quota allocated to a user.
  """
  cpu = float(cpu_str)
  ram = parse_data(ram_str).as_(Data.MB)
  disk = parse_data(disk_str).as_(Data.MB)

  client = make_admin_client_with_options(cluster)
  resp = client.get_quota(role)
  quota = resp.result.getQuotaResult.quota
  resource_details = ResourceManager.resource_details_from_quota(quota)
  log.info('Current quota for %s:\n\t%s' % (
      role,
      '\n\t'.join('%s\t%s%s' % (
          r.resource_type.display_name,
          r.value,
          r.resource_type.display_unit) for r in resource_details)))

  new_cpu = ResourceType.CPUS.value_type(
    cpu + ResourceManager.quantity_of(resource_details, ResourceType.CPUS))
  new_ram = ResourceType.RAM_MB.value_type(
    ram + ResourceManager.quantity_of(resource_details, ResourceType.RAM_MB))
  new_disk = ResourceType.DISK_MB.value_type(
    disk + ResourceManager.quantity_of(resource_details, ResourceType.DISK_MB))

  log.info('Attempting to update quota for %s to\n\tCPU\t%s\n\tRAM\t%s MB\n\tDisk\t%s MB' %
           (role, new_cpu, new_ram, new_disk))

  resp = client.set_quota(role, new_cpu, new_ram, new_disk)
  check_and_log_response(resp)

  def _maybe_update_health_check_count(self, is_healthy, reason):
    if not is_healthy:
      log.warning('Health check failure: %s' % reason)

      if self.current_consecutive_successes > 0:
        log.debug('Reset consecutive successes counter.')
        self.current_consecutive_successes = 0

      if self._should_ignore_failure():
        return

      if self._should_fail_fast():
        log.warning('Not enough attempts left prove health, failing fast.')
        self.healthy = False
        self.reason = reason

      self.current_consecutive_failures += 1
      if self.current_consecutive_failures > self.max_consecutive_failures:
        log.warning('Reached consecutive failure limit.')
        self.healthy = False
        self.reason = reason
    else:
      self.current_consecutive_successes += 1

      if not self.running:
        if self.current_consecutive_successes >= self.min_consecutive_successes:
          log.info('Reached consecutive success limit.')
          self.running = True

      if self.current_consecutive_failures > 0:
        log.debug('Reset consecutive failures counter.')
        self.current_consecutive_failures = 0

  def _update_instances_in_parallel(self, target, instances_to_update):
    """Processes instance updates in parallel and waits for completion.

    Arguments:
    target -- target method to handle instance update.
    instances_to_update -- list of InstanceData with update details.

    Returns Queue with non-updated instance data.
    """
    log.info('Processing in parallel with %s worker thread(s)' % self._update_config.batch_size)
    instance_queue = Queue()
    for instance_to_update in instances_to_update:
      instance_queue.put(instance_to_update)

    try:
      threads = []
      for _ in range(self._update_config.batch_size):
        threads.append(spawn_worker(target, kwargs={'instance_queue': instance_queue}))

      for thread in threads:
        thread.join_and_raise()
    except Exception:
      self._terminate()
      raise

    return instance_queue

  def add_member(self, service_instance):
    """
      Add the member to the ZooKeeper group.
      NOTE:
        - New members are slaves until being promoted.
        - A new member is not added if the specified service_instance already exists in the group.
      :return: The member ID for the ServiceInstance generated by ZooKeeper.
    """
    if not isinstance(service_instance, ServiceInstance):
      raise TypeError("'service_instance' should be a ServiceInstance")

    content = ServiceInstance.pack(service_instance)

    for k, v in self._cluster.members.items():
      if content == v:
        log.info("%s not added because it already exists in the group" % service_instance)
        return k

    znode_path = self._client.create(
        posixpath.join(self._cluster.slaves_group, self._cluster.MEMBER_PREFIX),
        content,
        sequence=True)
    _, member_id = posixpath.split(znode_path)
    with self._lock:
      self._cluster.members[member_id] = content
      return member_id

 def on_expiration(self):
   self._membership = None
   if not self._thread:
     return
   self._membership_termination = self._clock.time()
   log.info('Zookeeper session expired.')
   self.rejoin()

def select_binary_stream(base_path, version, name, config=None, url_opener=None):
  """Select a binary matching the current os and architecture.

  :param url_opener: Optional argument used only for testing, to 'pretend' to open urls.
  :returns: a 'stream' to download it from a support directory. The returned 'stream' is actually a
    lambda function which returns the files binary contents.
  :raises: :class:`pants.binary_util.BinaryUtil.BinaryNotFound` if no binary of the given version
    and name could not be found.
  """
  config = config or Config.load()
  baseurls = config.getdefault('pants_support_baseurls', type=list, default=[])
  if not baseurls:
    raise BinaryUtil.NoBaseUrlsError(
        'No urls are defined under pants_support_baseurls in the DEFAULT section of pants.ini.')
  timeout_secs = config.getdefault('pants_support_fetch_timeout_secs', type=int, default=30)
  binary_path = select_binary_base_path(base_path, version, name)
  if url_opener is None:
    url_opener = lambda u: closing(urllib_request.urlopen(u, timeout=timeout_secs))

  downloaded_successfully = False
  accumulated_errors = []
  for baseurl in OrderedSet(baseurls): # Wrap in OrderedSet because duplicates are wasteful.
    url = posixpath.join(baseurl, binary_path)
    log.info('Attempting to fetch {name} binary from: {url} ...'.format(name=name, url=url))
    try:
      with url_opener(url) as binary:
        log.info('Fetched {name} binary from: {url} .'.format(name=name, url=url))
        downloaded_successfully = True
        yield lambda: binary.read()
        break
    except (IOError, urllib_error.HTTPError, urllib_error.URLError, ValueError) as e:
      accumulated_errors.append('Failed to fetch binary from {url}: {error}'
                                .format(url=url, error=e))
  if not downloaded_successfully:
    raise BinaryUtil.BinaryNotFound((base_path, version, name), accumulated_errors)

def increase_quota(cluster, role, cpu_str, ram_str, disk_str):
    """usage: increase_quota cluster role cpu ram[unit] disk[unit]

  Increases the amount of production quota allocated to a user.
  """
    cpu = float(cpu_str)
    ram = parse_data(ram_str)
    disk = parse_data(disk_str)

    options = app.get_options()
    client = AuroraClientAPI(CLUSTERS[cluster], options.verbosity == "verbose")
    resp = client.get_quota(role)
    quota = resp.result.getQuotaResult.quota
    log.info(
        "Current quota for %s:\n\tCPU\t%s\n\tRAM\t%s MB\n\tDisk\t%s MB"
        % (role, quota.numCpus, quota.ramMb, quota.diskMb)
    )

    new_cpu = cpu + quota.numCpus
    new_ram = ram + Amount(quota.ramMb, Data.MB)
    new_disk = disk + Amount(quota.diskMb, Data.MB)

    log.info(
        "Attempting to update quota for %s to\n\tCPU\t%s\n\tRAM\t%s MB\n\tDisk\t%s MB"
        % (role, new_cpu, new_ram.as_(Data.MB), new_disk.as_(Data.MB))
    )

    resp = client.set_quota(role, new_cpu, new_ram.as_(Data.MB), new_disk.as_(Data.MB))
    check_and_log_response(resp)

    def this_is_really_our_pid(cls, process, current_user, start_time):
        """
      A heuristic to make sure that this is likely the pid that we own/forked.  Necessary
      because of pid-space wrapping.  We don't want to go and kill processes we don't own,
      especially if the killer is running as root.

      process: psutil.Process representing the process to check
      current_user: user expected to own the process
      start_time: time at which it's expected the process has started

      Raises:
        psutil.NoSuchProcess - if the Process supplied no longer exists
    """
        if process.username != current_user:
            log.info(
                "Expected pid %s to be ours but the pid user is %s and we're %s"
                % (process.pid, process.username, current_user)
            )
            return False

        if abs(start_time - process.create_time) >= cls.MAX_START_TIME_DRIFT.as_(Time.SECONDS):
            log.info("Expected pid %s start time to be %s but it's %s" % (process.pid, start_time, process.create_time))
            return False

        return True

def restore(j, target):
  """
  Restore jobs from a config directory
  """
  config_dir = app.get_options().config_dir

  if config_dir is None:
    log.error("no config_dir defined.")
    sys.exit()

  if not os.path.exists(os.path.realpath(config_dir)):
    log.error("config path does not exist")
    sys.exit()

  for job in os.listdir(config_dir):
    # here we need to:
    # check for config.xml
    # check for job on target server
    # if job exists, update it
    # if not create it. 
    config_file = "%s/%s/config.xml" % (config_dir, job)
    if not os.path.exists(config_file):
      log.error("config file does not exist: %s" %config_file)
      sys.exit()

    job_xml = read_config(config_file)

    try:
      jobj = j.get_job(job)
      if not jobj.get_config() == job_xml:
        log.info("Updating %s" % job)
        jobj.update_config(job_xml)
    except UnknownJob as e:
      log.error("job doesnt exist. creating")
      j.create_job(job, job_xml)

def handle_open(scheduler_url, role, env, job):
  url = synthesize_url(scheduler_url, role, env, job)
  if url:
    log.info('Job url: %s' % url)
    if app.get_options().open_browser:
      import webbrowser
      webbrowser.open_new_tab(url)

  def remove_cluster_state(self, cluster_name):
    path = self._get_cluster_state_path(cluster_name)
    if not os.path.isfile(path):
      log.info("No cluster state found on path %s" % path)
      return

    os.remove(path)

  def _resolve_image(cls, registry, name, tag, headers=None):
    url = MANIFESTS_URL % (registry, name, tag)
    response = requests.head(url, headers=headers)

    if response.status_code == requests.codes.unauthorized:
      # solve the auth challenge and retry again
      authorization = cls._solve_auth_challenge(response, registry)
      if headers is None:
        headers = dict()
      headers.update(authorization)
      response = requests.head(url, headers=headers)

      if response.status_code == requests.codes.unauthorized:
        # its a private repo, raise exception
        raise DockerClientException('Private Docker repository - %s:%s' % (name, tag))

    if response.status_code == requests.codes.ok:
      image_ref = '%[email protected]%s' % (name, response.headers.get('Docker-Content-Digest'))

      if registry != DEFAULT_DOCKER_REGISTRY_HOST:
        image_ref = '%s/%s' % (urlparse(registry).netloc, image_ref)

      log.info('Resolved %s:%s => %s' % (name, tag, image_ref))
      return image_ref

    # something is wrong
    response.raise_for_status()
    raise DockerClientException('Unable to resolve image %s:%s' % (name, tag))

  def _await_nailgun_server(self, workunit):
    nailgun_timeout_seconds = 5
    max_socket_connect_attempts = 10
    nailgun = None
    port_parse_start = time.time()
    with _safe_open(self._ng_out, 'r') as ng_out:
      while not nailgun:
        started = ng_out.readline()
        if started:
          port = self._parse_nailgun_port(started)
          with open(self._pidfile, 'a') as pidfile:
            pidfile.write(':%d\n' % port)
          nailgun = self._create_ngclient(port, workunit)
          log.debug('Detected ng server up on port %d' % port)
        elif time.time() - port_parse_start > nailgun_timeout_seconds:
          raise NailgunError('Failed to read ng output after %s seconds' % nailgun_timeout_seconds)

    attempt = 0
    while nailgun:
      sock = nailgun.try_connect()
      if sock:
        sock.close()
        log.info('Connected to ng server pid: %d @ port: %d' % self._get_nailgun_endpoint())
        return nailgun
      elif attempt > max_socket_connect_attempts:
        raise NailgunError('Failed to connect to ng output after %d connect attempts'
                            % max_socket_connect_attempts)
      attempt += 1
      log.debug('Failed to connect on attempt %d' % attempt)
      time.sleep(0.1)

    def wait_start(self, timeout=MAX_WAIT):
        log.debug("Waiting for task to start.")

        def is_started():
            return self._monitor and (self._monitor.active or self._monitor.finished)

        waited = Amount(0, Time.SECONDS)

        while waited < timeout:
            if not is_started():
                log.debug("  - sleeping...")
                self._clock.sleep(self.POLL_INTERVAL.as_(Time.SECONDS))
                waited += self.POLL_INTERVAL
            else:
                break

            if not self.is_alive:
                if self._popen_rc != 0:
                    raise TaskError("Task failed: %s" % self.compute_status().reason)
                else:
                    # We can end up here if the process exited between the call to Popen and
                    # waitpid (in is_alive), which is fine.
                    log.info("Task runner exited: %s" % self.compute_status().reason)
                    break

        if not is_started():
            log.error("Task did not start with in deadline, forcing loss.")
            self.lose()
            raise TaskError("Task did not start within deadline.")

    def _drain_hosts(self, drainable_hosts):
        """"Drains tasks from the specified hosts.

    This will move active tasks on these hosts to the DRAINING state, causing them to be
    rescheduled elsewhere.

    :param drainable_hosts: Hosts that are in maintenance mode and ready to be drained
    :type drainable_hosts: gen.apache.aurora.ttypes.Hosts
    :rtype: set of host names failed to drain
    """
        check_and_log_response(self._client.drain_hosts(drainable_hosts))
        drainable_hostnames = [hostname for hostname in drainable_hosts.hostNames]

        total_wait = self.STATUS_POLL_INTERVAL
        not_drained_hostnames = set(drainable_hostnames)
        while not self._wait_event.is_set() and not_drained_hostnames:
            log.info("Waiting for hosts to be in DRAINED: %s" % not_drained_hostnames)
            self._wait_event.wait(self.STATUS_POLL_INTERVAL.as_(Time.SECONDS))

            statuses = self.check_status(list(not_drained_hostnames))
            not_drained_hostnames = set(h[0] for h in statuses if h[1] != "DRAINED")

            total_wait += self.STATUS_POLL_INTERVAL
            if not_drained_hostnames and total_wait > self.MAX_STATUS_WAIT:
                log.warning(
                    "Failed to move all hosts into DRAINED within %s:\n%s"
                    % (
                        self.MAX_STATUS_WAIT,
                        "\n".join("\tHost:%s\tStatus:%s" % h for h in sorted(statuses) if h[1] != "DRAINED"),
                    )
                )
                break

        return not_drained_hostnames

  def promote_member(self, member_id):
    """
      Promote the member with the given ID to be the master of the cluster if it's not already the
      master.

      :return: True if the member is promoted. False if the member is already the master.
    """
    with self._lock:
      if member_id not in self._cluster.members:
        raise ValueError("Invalid member_id: %s" % member_id)

      # Do nothing if the member is already the master.
      if self._cluster.master and self._cluster.master == member_id:
        log.info("Not promoting %s because is already the master" % member_id)
        return False

      tx = self._client.transaction()
      if self._cluster.master:
        tx.delete(posixpath.join(self._cluster.master_group, self._cluster.master))
        self._cluster.members.pop(self._cluster.master)

      # "Move" the ZNode, i.e., create a ZNode of the same ID in the master group.
      tx.delete(posixpath.join(self._cluster.slaves_group, member_id))
      tx.create(
          posixpath.join(self._cluster.master_group, member_id),
          self._cluster.members[member_id])

      tx.commit()

      self._cluster.master = member_id

      return True

  def start(self, env=None):
    if self._process:
      log.warn("start() called when a running task subprocess already exists")
      return

    command = (
        "%(cmd)s %(framework_user)s %(host)s %(port)s %(server_id)s %(data_dir)s %(log_dir)s "
        "%(tmp_dir)s %(conf_file)s %(buffer_pool_size)s" % dict(
            cmd=os.path.join(self._scripts_dir, "mysos_launch_mysqld.sh"),
            framework_user=self._framework_user,
            host=self._host,
            port=self._port,
            server_id=self._server_id,
            data_dir=self._sandbox.mysql_data_dir,
            log_dir=self._sandbox.mysql_log_dir,
            tmp_dir=self._sandbox.mysql_tmp_dir,
            conf_file=self._conf_file,
            buffer_pool_size=self._buffer_pool_size))
    log.info("Executing command: %s" % command)
    self._process = subprocess.Popen(command, shell=True, env=env, preexec_fn=os.setpgrp)

    # There is a delay before mysqld becomes available to accept requests. Wait for it.
    command = "%(cmd)s %(pid_file)s %(port)s %(timeout)s" % dict(
        cmd=os.path.join(self._scripts_dir, "mysos_wait_for_mysqld.sh"),
        pid_file=os.path.join(self._sandbox.mysql_log_dir, "mysqld.pid"),
        port=self._port,
        timeout=60)
    log.info("Executing command: %s" % command)
    subprocess.check_call(command, shell=True, env=env)

    return self._process