Synnefo

def handle_message_delivery(func):

    """ Generic decorator for handling messages.

    This decorator is responsible for converting the message into json format,

    handling of common exceptions and acknowledment of message if needed.

            func(msg)

            client.basic_ack(message)

        except Exception as e:

            log.exception("Unexpected error: %s, msg: %s", e, msg)

    return wrapper

def instance_from_msg(func):

    """ Decorator for getting the VirtualMachine object of the msg.

    """

    @handle_message_delivery

    @wraps(func)

    def wrapper(msg):

        try:

            vm_id = utils.id_from_instance_name(msg["instance"])

            vm = VirtualMachine.objects.get(id=vm_id)

            func(vm, msg)

    return wrapper

def network_from_msg(func):

    """ Decorator for getting the Network object of the msg.

    """

    @handle_message_delivery

    @wraps(func)

    def wrapper(msg):

        try:

            network_id = utils.id_from_network_name(msg["network"])

            network = Network.objects.get(id=network_id)

            func(network, msg)

        except Network.InvalidBackendIdError:

            log.debug("Ignoring msg for unknown network %s.", msg['network'])

        except Network.DoesNotExist:

            log.error("Network %s with id %d not found in DB.",

                      msg['network'], vm_id)

    return wrapper

def if_update_required(func):

    """

    Decorator for checking if an incoming message needs to update the db.

    The database will not be updated in the following cases:

    - The message has been redelivered and the action has already been

      completed. In this case the event_time will be equal with the one

      in the database.

    - The message describes a previous state in the ganeti, from the one that is

      described in the db. In this case the event_time will be smaller from the

      one in the database.

    """

    @wraps(func)

    def wrapper(target, msg):

        event_time = merge_time(msg['event_time'])

        db_time = target.backendtime

        if event_time <= db_time:

            format_ = "%d/%m/%y %H:%M:%S:%f"

            log.debug("Ignoring message %s.\nevent_timestamp: %s db_timestamp: %s",

                      msg,

                      event_time.strftime(format_),

                      db_time.strftime(format_))

            return

        # New message. Update the database!

        func(target, msg, event_time)

@instance_from_msg

@if_update_required

def update_db(vm, msg, event_time):

@instance_from_msg

@if_update_required

def update_net(vm, msg, event_time):

@network_from_msg

@if_update_required

def update_network(network, msg, event_time):

    """Process a notification of type 'ganeti-network-status'"""

    log.debug("Processing ganeti-network-status msg: %s", msg)

    if msg['type'] != "ganeti-network-status":

        log.error("Message is of unknown type %s.", msg['type'])

        return

    log.debug("Done processing ganeti-network-status msg for vm %s.",

              msg['instance'])

@instance_from_msg

@if_update_required

def update_build_progress(vm, msg, event_time):

def dummy_proc(client, message, *args, **kwargs):

    QUEUE_GANETI_EVENTS_NETWORK = "%s-events-network" % prefix

    QUEUES = (QUEUE_GANETI_EVENTS_OP, QUEUE_GANETI_EVENTS_NETWORK, QUEUE_GANETI_EVENTS_NET, QUEUE_RECONC,

    # notifications of type "ganeti-network-status"

    DB_HANDLER_KEY_NETWORK = 'ganeti.%s.event.network' % prefix

    (QUEUE_GANETI_EVENTS_NETWORK, settings.EXCHANGE_GANETI, DB_HANDLER_KEY_NETWORK, 'update_network'),