Synnefo

# Copyright 2011 GRNET S.A. All rights reserved.

#

# Redistribution and use in source and binary forms, with or without

# modification, are permitted provided that the following conditions

# are met:

#

#   1. Redistributions of source code must retain the above copyright

#      notice, this list of conditions and the following disclaimer.

#

#  2. Redistributions in binary form must reproduce the above copyright

#     notice, this list of conditions and the following disclaimer in the

#     documentation and/or other materials provided with the distribution.

#

# THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND

# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE

# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE

# ARE DISCLAIMED.  IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE

# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL

# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS

# OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)

# HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT

# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY

# OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF

# SUCH DAMAGE.

#

# The views and conclusions contained in the software and documentation are

# those of the authors and should not be interpreted as representing official

# policies, either expressed or implied, of GRNET S.A.

# Callback functions used by the dispatcher to process incoming notifications

# from AMQP queues.

import logging

import json

from functools import wraps

from datetime import datetime

from synnefo.db.models import VirtualMachine

from synnefo.logic import utils, backend

from synnefo.lib.utils import merge_time

log = logging.getLogger()

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.

    """

    @wraps(func)

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

        try:

            msg = json.loads(message['body'])

            func(msg)

            client.basic_ack(message)

        except ValueError:

            log.error("Incoming message not in JSON format: %s", message)

            client.basic_ack(message)

        except KeyError:

            log.error("Malformed incoming JSON, missing attributes: %s",

                      message)

            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)

        except VirtualMachine.InvalidBackendIdError:

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

        except VirtualMachine.DoesNotExist:

            log.error("VM for instance %s with id %d not found in DB.",

                      msg['instance'], vm_id)

    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)

    return wrapper

@instance_from_msg

@if_update_required

def update_db(vm, msg, event_time):

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

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

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

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

        return

    backend.process_op_status(vm, event_time, msg['jobId'], msg['operation'],

                              msg['status'], msg['logmsg'])

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

              msg['instance'])

@instance_from_msg

@if_update_required

def update_net(vm, msg, event_time):

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

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

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

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

        return

    backend.process_net_status(vm, event_time, msg['nics'])

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

              msg["instance"])

@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):

    """Process a create progress message"""

    log.debug("Processing ganeti-create-progress msg: %s", msg)

    if msg['type'] != "ganeti-create-progress":

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

        return

    backend.process_create_progress(vm, event_time, msg['rprogress'], None)

    log.debug("Done processing ganeti-create-progress msg for vm %s.",

              msg['instance'])

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

    try:

        log.debug("Msg: %s", message['body'])

        client.basic_ack(message)

    except Exception as e:

        log.exception("Could not receive message")