def cast(name, message, service=False):
    logger.debug(
        "RPC cast to orchestrator:\n{0}".format(
            jsonutils.dumps(message, indent=4)
        )
    )
    #测试使用
    file_object = open('/opt/queuemsg.txt', 'w')
    file_object.write(jsonutils.dumps(message, indent=4))
    file_object.close()
    

    use_queue = naily_queue if not service else naily_service_queue
    use_exchange = naily_exchange if not service else naily_service_exchange
    with Connection(conn_str) as conn:
        with conn.Producer(serializer='json') as producer:
            publish = functools.partial(producer.publish, message,
                exchange=use_exchange, routing_key=name, declare=[use_queue])
            try:
                #pass
                publish()
            except amqp_exceptions.PreconditionFailed as e:
                logger.warning(six.text_type(e))
                # (dshulyak) we should drop both exchanges/queues in order
                # for astute to be able to recover temporary queues
                utils.delete_entities(
                    conn, naily_service_exchange, naily_service_queue,
                    naily_exchange, naily_queue)
                publish()

    def process_cluster_attributes(cls, cluster, attributes):
        """Generate Cluster-Plugins relation based on attributes.

        Iterates through plugins attributes, creates
        or deletes Cluster <-> Plugins relation if plugin
        is enabled or disabled.

        :param cluster: A cluster instance
        :type cluster: nailgun.db.sqlalchemy.models.cluster.Cluster
        :param attributes: Cluster attributes
        :type attributes: dict
        """
        plugins = {}

        # Detach plugins data
        for k in list(attributes):
            if cls.is_plugin_data(attributes[k]):
                plugins[k] = attributes.pop(k)['metadata']

        for container in six.itervalues(plugins):
            default = container.get('default', False)
            for attrs in container.get('versions', []):
                version_metadata = attrs.pop('metadata')
                plugin_id = version_metadata['plugin_id']
                plugin = Plugin.get_by_uid(plugin_id)
                if not plugin:
                    logger.warning(
                        'Plugin with id "%s" is not found, skip it', plugin_id)
                    continue
                enabled = container['enabled']\
                    and plugin_id == container['chosen_id']
                ClusterPlugin.set_attributes(
                    cluster.id, plugin.id, enabled=enabled,
                    attrs=attrs if enabled or default else None
                )

    def GET(self):
        """:returns: FUEL/FUELWeb commit SHA, release version.
        :http: * 200 (OK)
        """
        version = settings.VERSION
        method = settings.AUTH['AUTHENTICATION_METHOD']
        version['auth_required'] = method in ['fake', 'keystone']

        version['release_versions'] = {}
        for fl in glob.glob(self.release_versions):
            with open(fl, "r") as release_yaml:
                try:
                    version['release_versions'][
                        os.path.splitext(os.path.basename(fl))[0]
                    ] = yaml.load(
                        release_yaml.read()
                    )
                except Exception as exc:
                    logger.warning(
                        u"Failed to load release version "
                        "info from '{0}': {1}".format(
                            fl,
                            unicode(exc)
                        )
                    )

        return version

    def update_by_agent(cls, instance, data):
        """Update Node instance with some specific cases for agent.

        * don't update provisioning or error state back to discover
        * don't update volume information if disks arrays is empty

        :param data: dictionary of key-value pairs as object fields
        :returns: Node instance
        """
        # don't update provisioning and error back to discover
        data_status = data.get('status')
        if instance.status in ('provisioning', 'error'):
            if data.get('status', 'discover') == 'discover':
                logger.debug(
                    u"Node {0} has provisioning or error status - "
                    u"status not updated by agent".format(
                        instance.human_readable_name
                    )
                )

                data.pop('status', None)

        meta = data.get('meta', {})
        # don't update volume information, if agent has sent an empty array
        if len(meta.get('disks', [])) == 0 and instance.meta.get('disks'):

            logger.warning(
                u'Node {0} has received an empty disks array - '
                u'volume information will not be updated'.format(
                    instance.human_readable_name
                )
            )
            meta['disks'] = instance.meta['disks']

        # don't update volume information, if it is locked by node status
        if 'disks' in meta and cls.hardware_info_locked(instance):
            logger.debug("Volume information is locked for update on node %s",
                         instance.human_readable_name)
            meta['disks'] = instance.meta['disks']

        if not cls.is_interfaces_configuration_locked(instance) \
                and data.get('ip'):
            if instance.cluster_id:
                update_status = cls.check_ip_belongs_to_own_admin_network(
                    instance, data['ip'])
            else:
                update_status = cls.check_ip_belongs_to_any_admin_network(
                    instance, data['ip'])
            if update_status:
                if instance.status == consts.NODE_STATUSES.error and \
                        instance.error_type == consts.NODE_ERRORS.discover:
                    # accept the status from agent if the node had wrong IP
                    # previously
                    if data_status:
                        instance.status = data_status
                    else:
                        instance.status = consts.NODE_STATUSES.discover
            else:
                data.pop('status', None)
        return cls.update(instance, data)

 def PUT(self, node_id):
     node = self.get_object_or_404(Node, node_id)
     if not node.attributes:
         node.attributes = NodeAttributes(node_id=node.id)
     data = self.validator.validate_update(web.data())
     for key, value in data.iteritems():
         setattr(node, key, value)
         if key == 'cluster_id':
             if key:
                 self.allow_network_assignment_to_all_interfaces(node)
                 self.assign_networks_to_main_interface(node)
             else:
                 self.clear_assigned_networks(node)
                 self.clear_all_allowed_networks(node)
     if not node.status in ('provisioning', 'deploying') \
             and "role" in data or "cluster_id" in data:
         try:
             node.attributes.volumes = \
                 node.volume_manager.gen_volumes_info()
         except Exception as exc:
             msg = (
                 u"Failed to generate volumes "
                 "info for node '{0}': '{1}'"
             ).format(
                 node.name or data.get("mac") or data.get("id"),
                 str(exc) or "see logs for details"
             )
             logger.warning(traceback.format_exc())
             notifier.notify("error", msg, node_id=node.id)
     self.db.commit()
     return self.render(node)

    def update_pending_roles(cls, instance, new_pending_roles):
        if not instance.cluster_id:
            logger.warning(
                u"Attempting to assign pending roles to node "
                u"'{0}' which isn't added to cluster".format(
                    instance.name or instance.id
                )
            )
            return

        logger.debug(
            u"Updating pending roles for node {0}: {1}".format(
                instance.id,
                new_pending_roles
            )
        )

        if new_pending_roles == []:
            instance.pending_role_list = []
            # research why the hell we need this
            Cluster.clear_pending_changes(
                instance.cluster,
                node_id=instance.id
            )
        else:
            instance.pending_role_list = db().query(models.Role).filter_by(
                release_id=instance.cluster.release_id,
            ).filter(
                models.Role.name.in_(new_pending_roles)
            ).all()

        db().flush()
        db().refresh(instance)

    def update_volumes(cls, instance):
        attrs = instance.attributes
        if not attrs:
            attrs = cls.create_attributes(instance)

        try:
            attrs.volumes = instance.volume_manager.gen_volumes_info()
        except Exception as exc:
            msg = (
                u"Failed to generate volumes "
                u"info for node '{0}': '{1}'"
            ).format(
                instance.name or instance.mac or instance.id,
                str(exc) or "see logs for details"
            )
            logger.warning(traceback.format_exc())
            Notification.create({
                "topic": "error",
                "message": msg,
                "node_id": instance.id
            })

        if instance.cluster_id:
            Cluster.add_pending_changes(
                instance.cluster,
                "disks",
                node_id=instance.id
            )

        db().add(attrs)
        db().flush()

    def update_roles(cls, instance, new_roles):
        """Update roles for Node instance.
        Logs an error if node doesn't belong to Cluster

        :param instance: Node instance
        :param new_roles: list of new role names
        :returns: None
        """
        if not instance.cluster_id:
            logger.warning(
                u"Attempting to assign roles to node "
                u"'{0}' which isn't added to cluster".format(
                    instance.name or instance.id
                )
            )
            return

        if new_roles:
            instance.role_list = db().query(models.Role).filter_by(
                release_id=instance.cluster.release_id,
            ).filter(
                models.Role.name.in_(new_roles)
            ).all()
        else:
            instance.role_list = []
        db().flush()
        db().refresh(instance)

def set_proxy(proxy):
    """Replace http_proxy environment variable for the scope
    of context execution. After exit from context old proxy value
    (if any) is restored

    :param proxy: - proxy url
    """
    proxy_old_value = None

    if os.environ.get("http_proxy"):
        proxy_old_value = os.environ["http_proxy"]
        logger.warning("http_proxy variable is already set with "
                       "value: {0}. Change to {1}. Old value "
                       "will be restored after exit from script's "
                       "execution context"
                       .format(proxy_old_value, proxy))

    os.environ["http_proxy"] = proxy

    try:
        yield
    except Exception as e:
        logger.exception("Error while talking to proxy. Details: {0}"
                         .format(six.text_type(e)))
    finally:
        if proxy_old_value:
            logger.info("Restoring old value for http_proxy")
            os.environ["http_proxy"] = proxy_old_value
        else:
            logger.info("Deleting set http_proxy environment variable")
            del os.environ["http_proxy"]

    def get_nodes_by_role(cls, instance, role_name):
        """Get nodes related to some specific role

        :param instance: cluster db object
        :type: python object
        :param role_name: node role name
        :type: string
        """

        role = db().query(models.Role).filter_by(
            release_id=instance.release_id, name=role_name).first()

        if not role:
            logger.warning("%s role doesn't exist", role_name)
            return []

        nodes = db().query(models.Node).filter_by(cluster_id=instance.id)
        deployed_nodes = nodes.join(
            models.Node.role_list, aliased=True).filter(
                models.Role.id == role.id).all()
        pending_nodes = nodes.join(
            models.Node.pending_role_list, aliased=True).filter(
                models.Role.id == role.id).all()

        return deployed_nodes + pending_nodes

    def _update_dependencies(self):
        """Create dependencies that rely on regexp matching."""

        for task in six.itervalues(self.node):
            # tasks and groups should be used for declaring dependencies
            # between tasks and roles (which are simply group of tasks)
            available_groups = self.get_groups_subgraph().nodes()
            for group in task.get('groups', ()):
                pattern = NameMatchingPolicy.create(group)
                not_matched = []
                for available_group in available_groups:
                    if pattern.match(available_group):
                        self.add_edge(task['id'], available_group)
                    else:
                        not_matched.append(available_group)
                # Add dependency for non-existing group which will be
                # resolved in DeploymentGraphValidator
                if len(available_groups) == len(not_matched):
                    self.add_edge(task['id'], group)
                    logger.warning(
                        'Group "%s" is an invalid dependency', group)

                available_groups = not_matched

            for req in task.get('tasks', ()):
                self.add_edge(req, task['id'])

    def update_primary_roles(cls, instance, new_primary_roles):
        """Update primary_roles for Node instance.

        Logs an error if node doesn't belong to Cluster

        :param instance: Node instance
        :param new_primary_roles: list of new pending role names
        :returns: None
        """
        if not instance.cluster_id:
            logger.warning(
                u"Attempting to assign pending roles to node "
                u"'{0}' which isn't added to cluster".format(
                    instance.full_name))
            return

        assigned_roles = set(instance.roles + instance.pending_roles)
        for role in new_primary_roles:
            if role not in assigned_roles:
                logger.warning(
                    u"Could not mark node {0} as primary for {1} role, "
                    u"because there's no assigned {1} role.".format(
                        instance.full_name, role)
                )
                return

        logger.debug(
            u"Updating primary roles for node {0}: {1}".format(
                instance.full_name,
                new_primary_roles))

        instance.primary_roles = new_primary_roles
        db().flush()

    def _process_attr(cls, cluster, attr):
        if not isinstance(attr, dict):
            return

        metadata = attr.get('metadata', {})
        plugin_id = metadata.get('plugin_id')

        if not plugin_id:
            return

        plugin = Plugin.get_by_uid(plugin_id)
        if not plugin:
            logger.warning('Plugin with id "%s" is not found, skip it',
                           plugin_id)
            return

        enabled = metadata.get('enabled', False)

        # Value is true and plugin is not enabled for this cluster
        # that means plugin was enabled on this request
        if enabled and cluster not in plugin.clusters:
            plugin.clusters.append(cluster)
        # Value is false and plugin is enabled for this cluster
        # that means plugin was disabled on this request
        elif not enabled and cluster in plugin.clusters:
            plugin.clusters.remove(cluster)

    def POST(self, cluster_id):
        """:returns: Http response.
        :http: * 201 (nodes are successfully assigned)
               * 400 (invalid nodes data specified)
        """
        data = self.checked_data(
            self.validator.validate_collection_update,
            cluster_id=cluster_id
        )
        nodes = self.get_objects_list_or_404(Node, data.keys())
        cluster = self.get_object_or_404(Cluster, cluster_id)
        for node in nodes:
            node.cluster = cluster
            node.pending_roles = data[node.id]
            node.pending_addition = True
            try:
                node.attributes.volumes = \
                    node.volume_manager.gen_volumes_info()
                node.cluster.add_pending_changes("disks", node_id=node.id)

                network_manager = node.cluster.network_manager
                network_manager.assign_networks_by_default(node)
            except Exception as exc:
                logger.warning(traceback.format_exc())
                notifier.notify(
                    "error",
                    u"Failed to generate attributes for node '{0}': '{1}'"
                    .format(
                        node.human_readable_name(),
                        str(exc) or u"see logs for details"
                    ),
                    node_id=node.id
                )
            db().commit()
        raise web.ok

    def update_pending_roles(cls, instance, new_pending_roles):
        """Update pending_roles for Node instance.

        Logs an error if node doesn't belong to Cluster

        :param instance: Node instance
        :param new_pending_roles: list of new pending role names
        :returns: None
        """
        if not instance.cluster_id:
            logger.warning(
                u"Attempting to assign pending roles to node "
                u"'{0}' which isn't added to cluster".format(
                    instance.full_name))
            return

        logger.debug(
            u"Updating pending roles for node {0}: {1}".format(
                instance.full_name,
                new_pending_roles))

        if new_pending_roles == []:
            # TODO(enchantner): research why the hell we need this
            Cluster.clear_pending_changes(
                instance.cluster,
                node_id=instance.id
            )

        instance.pending_roles = new_pending_roles
        db().flush()

    def _get_pxe_iface_name(cls, node):
        """Returns appropriate pxe iface's name

        In case when node has network scheme configured
        we can not rely on its pxe interface calculation
        algorithm anymore, because admin ip is moving to
        bridge and 'pxe' property will have 'False' value
        for all interfaces. In this case we should rely on
        db where actual pxe interface was saved during the
        bootstrap stage.
        In case when node for some reason has no pxe interface
        in db we should get pxe interface using appropriate
        function `get_admin_physical_iface`.
        """
        db_interfaces = node.nic_interfaces
        pxe = next((
            i for i in node.meta['interfaces'] if i.get('pxe') or
                i.get('mac') == node.mac),
            None)
        if pxe:
            return pxe.get('name')
        pxe_db = next((i for i in db_interfaces if i.pxe), None)
        if pxe_db:
            return pxe_db.name
        if db_interfaces:
            return objects.Node.get_admin_physical_iface(node).name
        logger.warning(u'Cannot find pxe interface for node "%s"',
                       node.full_name)
        return None

    def create_cluster(self, api=True, exclude=None, **kwargs):
        cluster_data = {"name": "cluster-api-" + str(randint(0, 1000000))}
        editable_attributes = kwargs.pop("editable_attributes", None)

        if kwargs:
            cluster_data.update(kwargs)

        if "release_id" not in cluster_data:
            cluster_data["release_id"] = self.create_release(api=False).id

        if exclude and isinstance(exclude, list):
            for ex in exclude:
                try:
                    del cluster_data[ex]
                except KeyError as err:
                    logger.warning(err)
        if api:
            resp = self.app.post(
                reverse("ClusterCollectionHandler"),
                jsonutils.dumps(cluster_data),
                headers=self.default_headers,
                expect_errors=True,
            )
            self.tester.assertEqual(resp.status_code, 201, resp.body)
            cluster = resp.json_body
            cluster_db = Cluster.get_by_uid(cluster["id"])
        else:
            cluster = Cluster.create(cluster_data)
            cluster_db = cluster
            db().commit()
        self.clusters.append(cluster_db)

        if editable_attributes:
            Cluster.patch_attributes(cluster_db, {"editable": editable_attributes})
        return cluster

    def update_by_agent(cls, instance, data):
        """Update Node instance with some specific cases for agent.

        * don't update provisioning or error state back to discover
        * don't update volume information if disks arrays is empty

        :param data: dictionary of key-value pairs as object fields
        :returns: Node instance
        """
        # don't update provisioning and error back to discover
        if instance.status in ('provisioning', 'error'):
            if data.get('status', 'discover') == 'discover':
                logger.debug(
                    u"Node {0} has provisioning or error status - "
                    u"status not updated by agent".format(
                        instance.human_readable_name
                    )
                )

                data['status'] = instance.status

        # don't update volume information, if agent has sent an empty array
        meta = data.get('meta', {})
        if meta and len(meta.get('disks', [])) == 0 \
                and instance.meta.get('disks'):

            logger.warning(
                u'Node {0} has received an empty disks array - '
                u'volume information will not be updated'.format(
                    instance.human_readable_name
                )
            )
            meta['disks'] = instance.meta['disks']

        return cls.update(instance, data)

    def stop_deployment_resp(cls, **kwargs):
        logger.info("RPC method stop_deployment_resp received: %s" % jsonutils.dumps(kwargs))
        task_uuid = kwargs.get("task_uuid")
        nodes = kwargs.get("nodes", [])
        ia_nodes = kwargs.get("inaccessible_nodes", [])
        message = kwargs.get("error")
        status = kwargs.get("status")
        progress = kwargs.get("progress")

        task = objects.Task.get_by_uuid(task_uuid, fail_if_not_found=True)

        stopping_task_names = [consts.TASK_NAMES.deploy, consts.TASK_NAMES.deployment, consts.TASK_NAMES.provision]

        q_stop_tasks = objects.TaskCollection.filter_by_list(None, "name", stopping_task_names)
        q_stop_tasks = objects.TaskCollection.filter_by(q_stop_tasks, cluster_id=task.cluster_id)
        stop_tasks = objects.TaskCollection.order_by(q_stop_tasks, "id").all()

        # Locking cluster
        objects.Cluster.get_by_uid(task.cluster_id, fail_if_not_found=True, lock_for_update=True)

        if not stop_tasks:
            logger.warning(
                "stop_deployment_resp: deployment tasks \
                            not found for environment '%s'!",
                task.cluster_id,
            )

        if status == consts.TASK_STATUSES.ready:
            task.cluster.status = consts.CLUSTER_STATUSES.stopped

            if stop_tasks:
                map(db().delete, stop_tasks)

            node_uids = [n["uid"] for n in itertools.chain(nodes, ia_nodes)]
            q_nodes = objects.NodeCollection.filter_by_id_list(None, node_uids)
            q_nodes = objects.NodeCollection.filter_by(q_nodes, cluster_id=task.cluster_id)
            q_nodes = objects.NodeCollection.order_by(q_nodes, "id")
            q_nodes = objects.NodeCollection.lock_for_update(q_nodes)

            # locking Nodes for update
            update_nodes = objects.NodeCollection.lock_for_update(q_nodes).all()

            for node in update_nodes:
                objects.Node.reset_to_discover(node)

            if ia_nodes:
                cls._notify_inaccessible(task.cluster_id, [n["uid"] for n in ia_nodes], u"deployment stopping")

            message = (
                u"Deployment of environment '{0}' was successfully stopped. "
                u"Please make changes and reset the environment "
                u"if you want to redeploy it.".format(task.cluster.name or task.cluster_id)
            )

            notifier.notify("done", message, task.cluster_id)

        data = {"status": status, "progress": progress, "message": message}
        objects.Task.update(task, data)

        cls._update_action_log_entry(status, task.name, task_uuid, nodes)

 def run(self):
     while not self.stoprequest.isSet():
         self.db.expire_all()
         for node_db in self.db.query(Node).filter(
             # nodes may become unresponsive while provisioning
             not_(Node.status == 'provisioning')
         ):
             timedelta = (datetime.now() - node_db.timestamp).seconds
             if timedelta > self.timeout:
                 logger.warning(
                     u"Node '{0}' seems to be offline "
                     "for {1} seconds...".format(
                         node_db.name,
                         timedelta
                     )
                 )
                 if node_db.online:
                     node_db.online = False
                     self.db.add(node_db)
                     self.db.commit()
                     notifier.notify(
                         "error",
                         u"Node '{0}' has gone away".format(
                             node_db.name or node_db.mac
                         ),
                         node_id=node_db.id
                     )
         self.sleep()