# vim: tabstop=4 shiftwidth=4 softtabstop=4

# Copyright 2010 United States Government as represented by the
# Administrator of the National Aeronautics and Space Administration.
# All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

"""
Cloud Controller: Implementation of EC2 REST API calls, which are
dispatched to other nodes via AMQP RPC. State is via distributed
datastore.
"""

import base64
import json
import logging
import os
import time

from nova import vendor
from twisted.internet import defer

from nova import datastore
from nova import flags
from nova import rpc
from nova import utils
from nova import exception
from nova.auth import rbac
from nova.auth import users
from nova.compute import model
from nova.compute import network
from nova.compute import node
from nova.endpoint import images
from nova.volume import storage

FLAGS = flags.FLAGS

flags.DEFINE_string('cloud_topic', 'cloud', 'the topic clouds listen on')

def _gen_key(user_id, key_name):
    """ Tuck this into UserManager """
    try:
        manager = users.UserManager.instance()
        private_key, fingerprint = manager.generate_key_pair(user_id, key_name)
    except Exception as ex:
        return {'exception': ex}
    return {'private_key': private_key, 'fingerprint': fingerprint}


class CloudController(object):
    """ CloudController provides the critical dispatch between
 inbound API calls through the endpoint and messages
 sent to the other nodes.
"""
    def __init__(self):
        self.instdir = model.InstanceDirectory()
        self.network = network.PublicNetworkController()
        self.setup()

    @property
    def instances(self):
        """ All instances in the system, as dicts """
        return self.instdir.all

    @property
    def volumes(self):
        """ returns a list of all volumes """
        for volume_id in datastore.Redis.instance().smembers("volumes"):
            volume = storage.get_volume(volume_id)
            yield volume

    def __str__(self):
        return 'CloudController'

    def setup(self):
        """ Ensure the keychains and folders exist. """
        # Create keys folder, if it doesn't exist
        if not os.path.exists(FLAGS.keys_path):
            os.makedirs(os.path.abspath(FLAGS.keys_path))
        # Gen root CA, if we don't have one
        root_ca_path = os.path.join(FLAGS.ca_path, FLAGS.ca_file)
        if not os.path.exists(root_ca_path):
            start = os.getcwd()
            os.chdir(FLAGS.ca_path)
            utils.runthis("Generating root CA: %s", "sh genrootca.sh")
            os.chdir(start)
            # TODO: Do this with M2Crypto instead

    def get_instance_by_ip(self, ip):
        return self.instdir.by_ip(ip)

    def _get_mpi_data(self, project_id):
        result = {}
        for instance in self.instdir.all:
            if instance['project_id'] == project_id:
                line = '%s slots=%d' % (instance['private_dns_name'], node.INSTANCE_TYPES[instance['instance_type']]['vcpus'])
                if instance['key_name'] in result:
                    result[instance['key_name']].append(line)
                else:
                    result[instance['key_name']] = [line]
        return result

    def get_metadata(self, ip):
        i = self.get_instance_by_ip(ip)
        if i is None:
            return None
        mpi = self._get_mpi_data(i['project_id'])
        if i['key_name']:
            keys = {
                '0': {
                    '_name': i['key_name'],
                    'openssh-key': i['key_data']
                }
            }
        else:
            keys = ''
        data = {
            'user-data': base64.b64decode(i['user_data']),
            'meta-data': {
                'ami-id': i['image_id'],
                'ami-launch-index': i['ami_launch_index'],
                'ami-manifest-path': 'FIXME', # image property
                'block-device-mapping': { # TODO: replace with real data
                    'ami': 'sda1',
                    'ephemeral0': 'sda2',
                    'root': '/dev/sda1',
                    'swap': 'sda3'
                },
                'hostname': i['private_dns_name'], # is this public sometimes?
                'instance-action': 'none',
                'instance-id': i['instance_id'],
                'instance-type': i.get('instance_type', ''),
                'local-hostname': i['private_dns_name'],
                'local-ipv4': i['private_dns_name'], # TODO: switch to IP
                'kernel-id': i.get('kernel_id', ''),
                'placement': {
                    'availaibility-zone': i.get('availability_zone', 'nova'),
                },
                'public-hostname': i.get('dns_name', ''),
                'public-ipv4': i.get('dns_name', ''), # TODO: switch to IP
                'public-keys' : keys,
                'ramdisk-id': i.get('ramdisk_id', ''),
                'reservation-id': i['reservation_id'],
                'security-groups': i.get('groups', ''),
                'mpi': mpi
            }
        }
        if False: # TODO: store ancestor ids
            data['ancestor-ami-ids'] = []
        if i.get('product_codes', None):
            data['product-codes'] = i['product_codes']
        return data

    @rbac.allow('all')
    def describe_availability_zones(self, context, **kwargs):
        return {'availabilityZoneInfo': [{'zoneName': 'nova',
                                          'zoneState': 'available'}]}

    @rbac.allow('all')
    def describe_regions(self, context, region_name=None, **kwargs):
        # TODO(vish): region_name is an array.  Support filtering
        return {'regionInfo': [{'regionName': 'nova',
                                'regionUrl': FLAGS.ec2_url}]}

    @rbac.allow('all')
    def describe_snapshots(self,
                           context,
                           snapshot_id=None,
                           owner=None,
                           restorable_by=None,
                           **kwargs):
        return {'snapshotSet': [{'snapshotId': 'fixme',
                                 'volumeId': 'fixme',
                                 'status': 'fixme',
                                 'startTime': 'fixme',
                                 'progress': 'fixme',
                                 'ownerId': 'fixme',
                                 'volumeSize': 0,
                                 'description': 'fixme'}]}

    @rbac.allow('all')
    def describe_key_pairs(self, context, key_name=None, **kwargs):
        key_pairs = context.user.get_key_pairs()
        if not key_name is None:
            key_pairs = [x for x in key_pairs if x.name in key_name]

        result = []
        for key_pair in key_pairs:
            # filter out the vpn keys
            suffix = FLAGS.vpn_key_suffix
            if context.user.is_admin() or not key_pair.name.endswith(suffix):
                result.append({
                    'keyName': key_pair.name,
                    'keyFingerprint': key_pair.fingerprint,
                })

        return { 'keypairsSet': result }

    @rbac.allow('all')
    def create_key_pair(self, context, key_name, **kwargs):
        try:
            d = defer.Deferred()
            p = context.handler.application.settings.get('pool')
            def _complete(kwargs):
                if 'exception' in kwargs:
                    d.errback(kwargs['exception'])
                    return
                d.callback({'keyName': key_name,
                    'keyFingerprint': kwargs['fingerprint'],
                    'keyMaterial': kwargs['private_key']})
            p.apply_async(_gen_key, [context.user.id, key_name],
                callback=_complete)
            return d

        except users.UserError, e:
            raise

    @rbac.allow('all')
    def delete_key_pair(self, context, key_name, **kwargs):
        context.user.delete_key_pair(key_name)
        # aws returns true even if the key doens't exist
        return True

    @rbac.allow('all')
    def describe_security_groups(self, context, group_names, **kwargs):
        groups = { 'securityGroupSet': [] }

        # Stubbed for now to unblock other things.
        return groups

    @rbac.allow('netadmin')
    def create_security_group(self, context, group_name, **kwargs):
        return True

    @rbac.allow('netadmin')
    def delete_security_group(self, context, group_name, **kwargs):
        return True

    @rbac.allow('projectmanager', 'sysadmin')
    def get_console_output(self, context, instance_id, **kwargs):
        # instance_id is passed in as a list of instances
        instance = self._get_instance(context, instance_id[0])
        return rpc.call('%s.%s' % (FLAGS.compute_topic, instance['node_name']),
            {"method": "get_console_output",
             "args" : {"instance_id": instance_id[0]}})

    def _get_user_id(self, context):
        if context and context.user:
            return context.user.id
        else:
            return None

    @rbac.allow('projectmanager', 'sysadmin')
    def describe_volumes(self, context, **kwargs):
        volumes = []
        for volume in self.volumes:
            if context.user.is_admin() or volume['project_id'] == context.project.id:
                v = self.format_volume(context, volume)
                volumes.append(v)
        return defer.succeed({'volumeSet': volumes})

    def format_volume(self, context, volume):
        v = {}
        v['volumeId'] = volume['volume_id']
        v['status'] = volume['status']
        v['size'] = volume['size']
        v['availabilityZone'] = volume['availability_zone']
        v['createTime'] = volume['create_time']
        if context.user.is_admin():
            v['status'] = '%s (%s, %s, %s, %s)' % (
                volume.get('status', None),
                volume.get('user_id', None),
                volume.get('node_name', None),
                volume.get('instance_id', ''),
                volume.get('mountpoint', ''))
        if volume['attach_status'] == 'attached':
            v['attachmentSet'] = [{'attachTime': volume['attach_time'],
                                   'deleteOnTermination': volume['delete_on_termination'],
                                   'device' : volume['mountpoint'],
                                   'instanceId' : volume['instance_id'],
                                   'status' : 'attached',
                                   'volume_id' : volume['volume_id']}]
        else:
            v['attachmentSet'] = [{}]
        return v

    @rbac.allow('projectmanager', 'sysadmin')
    def create_volume(self, context, size, **kwargs):
        # TODO(vish): refactor this to create the volume object here and tell storage to create it
        res = rpc.call(FLAGS.storage_topic, {"method": "create_volume",
                                 "args" : {"size": size,
                                           "user_id": context.user.id,
                                           "project_id": context.project.id}})
        def _format_result(result):
            volume = self._get_volume(context, result['result'])
            return {'volumeSet': [self.format_volume(context, volume)]}
        res.addCallback(_format_result)
        return res

    def _get_address(self, context, public_ip):
        # FIXME(vish) this should move into network.py
        address = self.network.get_host(public_ip)
        if address and (context.user.is_admin() or address['project_id'] == context.project.id):
            return address
        raise exception.NotFound("Address at ip %s not found" % public_ip)

    def _get_image(self, context, image_id):
        """passes in context because
        objectstore does its own authorization"""
        result = images.list(context, [image_id])
        if not result:
            raise exception.NotFound('Image %s could not be found' % image_id)
        image = result[0]
        return image

    def _get_instance(self, context, instance_id):
        for instance in self.instdir.all:
            if instance['instance_id'] == instance_id:
                if context.user.is_admin() or instance['project_id'] == context.project.id:
                    return instance
        raise exception.NotFound('Instance %s could not be found' % instance_id)

    def _get_volume(self, context, volume_id):
        volume = storage.get_volume(volume_id)
        if context.user.is_admin() or volume['project_id'] == context.project.id:
            return volume
        raise exception.NotFound('Volume %s could not be found' % volume_id)

    @rbac.allow('projectmanager', 'sysadmin')
    def attach_volume(self, context, volume_id, instance_id, device, **kwargs):
        volume = self._get_volume(context, volume_id)
        if volume['status'] == "attached":
            raise exception.ApiError("Volume is already attached")
        # TODO(vish): looping through all volumes is slow. We should probably maintain an index
        for vol in self.volumes:
            if vol['instance_id'] == instance_id and vol['mountpoint'] == device:
                raise exception.ApiError("Volume %s is already attached to %s" % (vol['volume_id'], vol['mountpoint']))
        volume.start_attach(instance_id, device)
        instance = self._get_instance(context, instance_id)
        compute_node = instance['node_name']
        rpc.cast('%s.%s' % (FLAGS.compute_topic, compute_node),
                                {"method": "attach_volume",
                                 "args" : {"volume_id": volume_id,
                                           "instance_id" : instance_id,
                                           "mountpoint" : device}})
        return defer.succeed({'attachTime' : volume['attach_time'],
                              'device' : volume['mountpoint'],
                              'instanceId' : instance_id,
                              'requestId' : context.request_id,
                              'status' : volume['attach_status'],
                              'volumeId' : volume_id})


    @rbac.allow('projectmanager', 'sysadmin')
    def detach_volume(self, context, volume_id, **kwargs):
        volume = self._get_volume(context, volume_id)
        instance_id = volume.get('instance_id', None)
        if not instance_id:
            raise exception.Error("Volume isn't attached to anything!")
        if volume['status'] == "available":
            raise exception.Error("Volume is already detached")
        try:
            volume.start_detach()
            instance = self._get_instance(context, instance_id)
            rpc.cast('%s.%s' % (FLAGS.compute_topic, instance['node_name']),
                                {"method": "detach_volume",
                                 "args" : {"instance_id": instance_id,
                                           "volume_id": volume_id}})
        except exception.NotFound:
            # If the instance doesn't exist anymore,
            # then we need to call detach blind
            volume.finish_detach()
        return defer.succeed({'attachTime' : volume['attach_time'],
                              'device' : volume['mountpoint'],
                              'instanceId' : instance_id,
                              'requestId' : context.request_id,
                              'status' : volume['attach_status'],
                              'volumeId' : volume_id})

    def _convert_to_set(self, lst, label):
        if lst == None or lst == []:
            return None
        if not isinstance(lst, list):
            lst = [lst]
        return [{label: x} for x in lst]

    @rbac.allow('all')
    def describe_instances(self, context, **kwargs):
        return defer.succeed(self._format_instances(context))

    def _format_instances(self, context, reservation_id = None):
        reservations = {}
        if context.user.is_admin():
            instgenerator = self.instdir.all
        else:
            instgenerator = self.instdir.by_project(context.project.id)
        for instance in instgenerator:
            res_id = instance.get('reservation_id', 'Unknown')
            if reservation_id != None and reservation_id != res_id:
                continue
            if not context.user.is_admin():
                if instance['image_id'] == FLAGS.vpn_image_id:
                    continue
            i = {}
            i['instance_id'] = instance.get('instance_id', None)
            i['image_id'] = instance.get('image_id', None)
            i['instance_state'] = {
                'code': instance.get('state', 0),
                'name': instance.get('state_description', 'pending')
            }
            i['public_dns_name'] = self.network.get_public_ip_for_instance(
                                                        i['instance_id'])
            i['private_dns_name'] = instance.get('private_dns_name', None)
            if not i['public_dns_name']:
                i['public_dns_name'] = i['private_dns_name']
            i['dns_name'] = instance.get('dns_name', None)
            i['key_name'] = instance.get('key_name', None)
            if context.user.is_admin():
                i['key_name'] = '%s (%s, %s)' % (i['key_name'],
                    instance.get('project_id', None), instance.get('node_name',''))
            i['product_codes_set'] = self._convert_to_set(
                instance.get('product_codes', None), 'product_code')
            i['instance_type'] = instance.get('instance_type', None)
            i['launch_time'] = instance.get('launch_time', None)
            i['ami_launch_index'] = instance.get('ami_launch_index',
                                                 None)
            if not reservations.has_key(res_id):
                r = {}
                r['reservation_id'] = res_id
                r['owner_id'] = instance.get('project_id', None)
                r['group_set'] = self._convert_to_set(
                    instance.get('groups', None), 'group_id')
                r['instances_set'] = []
                reservations[res_id] = r
            reservations[res_id]['instances_set'].append(i)

        instance_response = {'reservationSet' : list(reservations.values()) }
        return instance_response

    @rbac.allow('all')
    def describe_addresses(self, context, **kwargs):
        return self.format_addresses(context)

    def format_addresses(self, context):
        addresses = []
        # TODO(vish): move authorization checking into network.py
        for address in self.network.host_objs:
            #logging.debug(address_record)
            address_rv = {
                'public_ip': address['address'],
                'instance_id' : address.get('instance_id', 'free')
            }
            if context.user.is_admin():
                address_rv['instance_id'] = "%s (%s, %s)" % (
                    address['instance_id'],
                    address['user_id'],
                    address['project_id'],
                )
            addresses.append(address_rv)
        # logging.debug(addresses)
        return {'addressesSet': addresses}

    @rbac.allow('netadmin')
    def allocate_address(self, context, **kwargs):
        address = self.network.allocate_ip(
                                context.user.id, context.project.id, 'public')
        return defer.succeed({'addressSet': [{'publicIp' : address}]})

    @rbac.allow('netadmin')
    def release_address(self, context, public_ip, **kwargs):
        self.network.deallocate_ip(public_ip)
        return defer.succeed({'releaseResponse': ["Address released."]})

    @rbac.allow('netadmin')
    def associate_address(self, context, instance_id, **kwargs):
        instance = self._get_instance(context, instance_id)
        self.network.associate_address(
                            kwargs['public_ip'],
                            instance['private_dns_name'],
                            instance_id)
        return defer.succeed({'associateResponse': ["Address associated."]})

    @rbac.allow('netadmin')
    def disassociate_address(self, context, public_ip, **kwargs):
        address = self._get_address(context, public_ip)
        self.network.disassociate_address(public_ip)
        # TODO - Strip the IP from the instance
        return defer.succeed({'disassociateResponse': ["Address disassociated."]})

    def release_ip(self, context, private_ip, **kwargs):
        self.network.release_ip(private_ip)
        return defer.succeed({'releaseResponse': ["Address released."]})

    def lease_ip(self, context, private_ip, **kwargs):
        self.network.lease_ip(private_ip)
        return defer.succeed({'leaseResponse': ["Address leased."]})

    @rbac.allow('projectmanager', 'sysadmin')
    def run_instances(self, context, **kwargs):
        # make sure user can access the image
        # vpn image is private so it doesn't show up on lists
        if kwargs['image_id'] != FLAGS.vpn_image_id:
            image = self._get_image(context, kwargs['image_id'])

        # FIXME(ja): if image is cloudpipe, this breaks

        # get defaults from imagestore
        image_id = image['imageId']
        kernel_id = image.get('kernelId', None)
        ramdisk_id = image.get('ramdiskId', None)

        # API parameters overrides of defaults
        kernel_id = kwargs.get('kernel_id', kernel_id)
        ramdisk_id = kwargs.get('ramdisk_id', ramdisk_id)

        logging.debug("Going to run instances...")
        reservation_id = utils.generate_uid('r')
        launch_time = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())
        key_data = None
        if kwargs.has_key('key_name'):
            key_pair = context.user.get_key_pair(kwargs['key_name'])
            if not key_pair:
                raise exception.ApiError('Key Pair %s not found' %
                                         kwargs['key_name'])
            key_data = key_pair.public_key
        # TODO: Get the real security group of launch in here
        security_group = "default"
        if FLAGS.simple_network:
            bridge_name = FLAGS.simple_network_bridge
        else:
            net = network.BridgedNetwork.get_network_for_project(
                    context.user.id, context.project.id, security_group)
            bridge_name = net['bridge_name']
        for num in range(int(kwargs['max_count'])):
            inst = self.instdir.new()
            inst['image_id'] = image_id
            inst['kernel_id'] = kernel_id
            inst['ramdisk_id'] = ramdisk_id
            inst['user_data'] = kwargs.get('user_data', '')
            inst['instance_type'] = kwargs.get('instance_type', 'm1.small')
            inst['reservation_id'] = reservation_id
            inst['launch_time'] = launch_time
            inst['key_data'] = key_data or ''
            inst['key_name'] = kwargs.get('key_name', '')
            inst['user_id'] = context.user.id
            inst['project_id'] = context.project.id
            inst['mac_address'] = utils.generate_mac()
            inst['ami_launch_index'] = num
            inst['bridge_name'] = bridge_name
            if FLAGS.simple_network:
                address = network.allocate_simple_ip()
            else:
                if inst['image_id'] == FLAGS.vpn_image_id:
                    address = network.allocate_vpn_ip(
                            inst['user_id'],
                            inst['project_id'],
                            mac=inst['mac_address'])
                else:
                    address = network.allocate_ip(
                            inst['user_id'],
                            inst['project_id'],
                            mac=inst['mac_address'])
            inst['private_dns_name'] = str(address)
            # TODO: allocate expresses on the router node
            inst.save()
            rpc.cast(FLAGS.compute_topic,
                 {"method": "run_instance",
                  "args": {"instance_id" : inst.instance_id}})
            logging.debug("Casting to node for %s's instance with IP of %s" %
                      (context.user.name, inst['private_dns_name']))
        # TODO: Make the NetworkComputeNode figure out the network name from ip.
        return defer.succeed(self._format_instances(
                                context, reservation_id))

    @rbac.allow('projectmanager', 'sysadmin')
    def terminate_instances(self, context, instance_id, **kwargs):
        logging.debug("Going to start terminating instances")
        for i in instance_id:
            logging.debug("Going to try and terminate %s" % i)
            try:
                instance = self._get_instance(context, i)
            except exception.NotFound:
                logging.warning("Instance %s was not found during terminate" % i)
                continue
            try:
                self.network.disassociate_address(
                    instance.get('public_dns_name', 'bork'))
            except:
                pass
            if instance.get('private_dns_name', None):
                logging.debug("Deallocating address %s" % instance.get('private_dns_name', None))
                if FLAGS.simple_network:
                    network.deallocate_simple_ip(instance.get('private_dns_name', None))
                else:
                    try:
                        self.network.deallocate_ip(instance.get('private_dns_name', None))
                    except Exception, _err:
                        pass
            if instance.get('node_name', 'unassigned') != 'unassigned':  #It's also internal default
                rpc.cast('%s.%s' % (FLAGS.compute_topic, instance['node_name']),
                             {"method": "terminate_instance",
                              "args" : {"instance_id": i}})
            else:
                instance.destroy()
        return defer.succeed(True)

    @rbac.allow('projectmanager', 'sysadmin')
    def reboot_instances(self, context, instance_id, **kwargs):
        """instance_id is a list of instance ids"""
        for i in instance_id:
            instance = self._get_instance(context, i)
            rpc.cast('%s.%s' % (FLAGS.compute_topic, instance['node_name']),
                             {"method": "reboot_instance",
                              "args" : {"instance_id": i}})
        return defer.succeed(True)

    @rbac.allow('projectmanager', 'sysadmin')
    def delete_volume(self, context, volume_id, **kwargs):
        # TODO: return error if not authorized
        volume = self._get_volume(context, volume_id)
        storage_node = volume['node_name']
        rpc.cast('%s.%s' % (FLAGS.storage_topic, storage_node),
                            {"method": "delete_volume",
                             "args" : {"volume_id": volume_id}})
        return defer.succeed(True)

    @rbac.allow('all')
    def describe_images(self, context, image_id=None, **kwargs):
        # The objectstore does its own authorization for describe
        imageSet = images.list(context, image_id)
        return defer.succeed({'imagesSet': imageSet})

    @rbac.allow('projectmanager', 'sysadmin')
    def deregister_image(self, context, image_id, **kwargs):
        # FIXME: should the objectstore be doing these authorization checks?
        images.deregister(context, image_id)
        return defer.succeed({'imageId': image_id})

    @rbac.allow('projectmanager', 'sysadmin')
    def register_image(self, context, image_location=None, **kwargs):
        # FIXME: should the objectstore be doing these authorization checks?
        if image_location is None and kwargs.has_key('name'):
            image_location = kwargs['name']
        image_id = images.register(context, image_location)
        logging.debug("Registered %s as %s" % (image_location, image_id))

        return defer.succeed({'imageId': image_id})

    @rbac.allow('all')
    def describe_image_attribute(self, context, image_id, attribute, **kwargs):
        if attribute != 'launchPermission':
            raise exception.ApiError('attribute not supported: %s' % attribute)
        try:
            image = images.list(context, image_id)[0]
        except IndexError:
            raise exception.ApiError('invalid id: %s' % image_id)
        result = { 'image_id': image_id, 'launchPermission': [] }
        if image['isPublic']:
            result['launchPermission'].append({ 'group': 'all' })
        return defer.succeed(result)

    @rbac.allow('projectmanager', 'sysadmin')
    def modify_image_attribute(self, context, image_id, attribute, operation_type, **kwargs):
        # TODO(devcamcar): Support users and groups other than 'all'.
        if attribute != 'launchPermission':
            raise exception.ApiError('attribute not supported: %s' % attribute)
        if len(kwargs['user_group']) != 1 and kwargs['user_group'][0] != 'all':
            raise exception.ApiError('only group "all" is supported')
        if not operation_type in ['add', 'remove']:
            raise exception.ApiError('operation_type must be add or remove')
        result = images.modify(context, image_id, operation_type)
        return defer.succeed(result)

    def update_state(self, topic, value):
        """ accepts status reports from the queue and consolidates them """
        # TODO(jmc): if an instance has disappeared from
        # the node, call instance_death
        if topic == "instances":
            return defer.succeed(True)
        aggregate_state = getattr(self, topic)
        node_name = value.keys()[0]
        items = value[node_name]

        logging.debug("Updating %s state for %s" % (topic, node_name))

        for item_id in items.keys():
            if (aggregate_state.has_key('pending') and
                aggregate_state['pending'].has_key(item_id)):
                del aggregate_state['pending'][item_id]
        aggregate_state[node_name] = items

        return defer.succeed(True)
