ucloud-scheduler/main.py

# TODO
#  1. send an email to an email address defined by env['admin-email']
#     if resources are finished
#  2. Introduce a status endpoint of the scheduler -
#     maybe expose a prometheus compatible output

import logging

from config import etcd_client as client
from config import (host_pool, request_pool, vm_pool, request_prefix)
from ucloud_common.request import RequestEntry, RequestType
from helper import (get_suitable_host, dead_host_mitigation, dead_host_detection,
                    assign_host, NoSuitableHostFound)

pending_vms = []


def main():
    global pending_vms

    for request_iterator in [
        client.get_prefix(request_prefix, value_in_json=True),
        client.watch_prefix(request_prefix, timeout=5, value_in_json=True),
    ]:
        for request_event in request_iterator:
            request_entry = RequestEntry(request_event)
            logging.debug("{}, {}".format(request_entry.key, request_entry.value))

            # Never Run time critical mechanism inside timeout
            # mechanism because timeout mechanism only comes
            # when no other event is happening. It means under
            # heavy load there would not be a timeout event.
            if request_entry._type == "TIMEOUT":

                # Detect hosts that are dead and set their status
                # to "DEAD", and their VMs' status to "KILLED"
                logging.debug("TIMEOUT event occured")
                dead_hosts = dead_host_detection()
                logging.debug("Dead hosts: {}".format(dead_hosts))
                dead_host_mitigation(dead_hosts)

                # If there are VMs that weren't assigned a host
                # because there wasn't a host available which
                # meets requirement of that VM then we would
                # create a new ScheduleVM request for that VM
                # on our behalf.
                while pending_vms:
                    pending_vm_entry = pending_vms.pop()
                    r = RequestEntry.from_scratch(type="ScheduleVM",
                                                  uuid=pending_vm_entry.uuid,
                                                  hostname=pending_vm_entry.hostname)
                    request_pool.put(r)

            elif request_entry._type == RequestType.ScheduleVM:
                vm_entry = vm_pool.get(request_entry.uuid)
                client.client.delete(request_entry.key)  # consume Request

                # If the Request is about a VM which is labelled as "migration"
                # and has a destination
                if hasattr(request_entry, "migration") and request_entry.migration \
                        and hasattr(request_entry, "destination") and request_entry.destination:
                    try:
                        get_suitable_host(vm_entry.specs, [host_pool.get(request_entry.destination)])
                    except NoSuitableHostFound:
                        logging.info("Requested destination host doesn't have enough capacity"
                                     "to hold {}".format(vm_entry.uuid))
                    else:
                        r = RequestEntry.from_scratch(type=RequestType.InitVMMigration,
                                                      uuid=request_entry.uuid,
                                                      destination=request_entry.destination)
                        request_pool.put(r)

                # If the Request is about a VM that just want to get started/created
                else:
                    # assign_host only returns None when we couldn't be able to assign
                    # a host to a VM because of resource constraints
                    try:
                        assign_host(vm_entry)
                    except NoSuitableHostFound:
                        vm_entry.log.append("Can't schedule VM. No Resource Left.")
                        vm_pool.put(vm_entry)

                        pending_vms.append(vm_entry)
                        logging.info("No Resource Left. Emailing admin....")


logging.info("{} SESSION STARTED {}".format('*' * 5, '*' * 5))
main()