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 stats_user_resp(cls, **kwargs):
        logger.info("RPC method stats_user_resp received: %s",
                    jsonutils.dumps(kwargs))

        task_uuid = kwargs.get('task_uuid')
        nodes = kwargs.get('nodes', [])
        status = kwargs.get('status')
        error = kwargs.get('error')
        message = kwargs.get('msg')

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

        if status not in (consts.TASK_STATUSES.ready,
                          consts.TASK_STATUSES.error):
            logger.debug("Task %s, id: %s in status: %s",
                         task.name, task.id, task.status)
            return

        data = {'status': status, 'progress': 100, 'message': message}
        if status == consts.TASK_STATUSES.error:
            logger.error("Task %s, id: %s failed: %s",
                         task.name, task.id, error)
            data['message'] = error

        objects.Task.update(task, data)
        cls._update_action_log_entry(status, task.name, task_uuid, nodes)
        logger.info("RPC method stats_user_resp processed")

    def message(cls, task, nodes):
        logger.debug("%s.message(task=%s)", cls.__class__.__name__, task.uuid)

        for n in nodes:
            if n.pending_roles:
                n.roles += n.pending_roles
                n.pending_roles = []
            n.status = 'provisioned'
            n.progress = 0

        # here we replace deployment data if user redefined them
        serialized_cluster = deployment_serializers.serialize(
            task.cluster, nodes)

        # After serialization set pending_addition to False
        for node in nodes:
            node.pending_addition = False

        rpc_message = make_astute_message(
            task,
            'deploy',
            'deploy_resp',
            {
                'deployment_info': serialized_cluster
            }
        )
        db().commit()
        return rpc_message

 def update_slave_nodes_fqdn(cls, nodes):
     for n in nodes:
         fqdn = cls.make_slave_fqdn(n.id)
         if n.fqdn != fqdn:
             n.fqdn = fqdn
             logger.debug("Updating node fqdn: %s %s", n.id, n.fqdn)
             db().commit()

    def assign_admin_ips(cls, node_id, num=1):
        """Method for assigning admin IP addresses to nodes.

        :param node_id: Node database ID.
        :type  node_id: int
        :param num: Number of IP addresses for node.
        :type  num: int
        :returns: None
        """
        admin_net_id = cls.get_admin_network_group_id()
        node_admin_ips = db().query(IPAddr).filter_by(
            node=node_id,
            network=admin_net_id
        ).all()

        if not node_admin_ips or len(node_admin_ips) < num:
            admin_net = db().query(NetworkGroup).get(admin_net_id)
            logger.debug(
                u"Trying to assign admin ips: node=%s count=%s",
                node_id,
                num - len(node_admin_ips)
            )
            free_ips = cls.get_free_ips(
                admin_net.id,
                num=num - len(node_admin_ips)
            )
            logger.info(len(free_ips))
            for ip in free_ips:
                ip_db = IPAddr(
                    node=node_id,
                    ip_addr=ip,
                    network=admin_net_id
                )
                db().add(ip_db)
            db().commit()

    def update_verify_networks(cls, uuid, status,
                               progress, msg, result):
        #TODO(dshulyak) move network tests into ostf
        task = db().query(Task).filter_by(uuid=uuid).first()
        if not task:
            logger.error("Can't set status='%s', message='%s': No task \
                    with UUID %s found!", status, msg, uuid)
            return

        previous_status = task.status

        statuses = [sub.status for sub in task.subtasks]
        messages = [sub.message for sub in task.subtasks]
        messages.append(msg)
        statuses.append(status)
        if any(st == 'error' for st in statuses):
            task.status = 'error'
        else:
            task.status = status or task.status
        task.progress = progress or task.progress
        task.result = result or task.result
        # join messages if not None or ""
        task.message = '\n'.join([m for m in messages if m])
        db().commit()
        if previous_status != task.status and task.cluster_id:
            logger.debug("Updating cluster status: "
                         "cluster_id: %s status: %s",
                         task.cluster_id, status)
            cls.update_cluster_status(uuid)

    def PUT(self, cluster_id):
        """:returns: JSONized Cluster attributes.

        :http: * 200 (OK)
               * 400 (wrong attributes data specified)
               * 404 (cluster not found in db)
               * 500 (cluster has no attributes)
        """
        cluster = self.get_object_or_404(
            objects.Cluster,
            cluster_id,
            log_404=(
                "error",
                "There is no cluster "
                "with id '{0}' in DB.".format(cluster_id)
            )
        )

        if not cluster.attributes:
            logger.error('ClusterAttributesDefaultsHandler: no attributes'
                         ' found for cluster_id %s' % cluster_id)
            raise self.http(500, "No attributes found!")

        cluster.attributes.editable = (
            objects.Cluster.get_default_editable_attributes(cluster))
        objects.Cluster.add_pending_changes(cluster, "attributes")

        logger.debug('ClusterAttributesDefaultsHandler:'
                     ' editable attributes for cluster_id %s were reset'
                     ' to default' % cluster_id)
        return {"editable": cluster.attributes.editable}

 def execute(self):
     logger.debug("Creating release dowload task")
     task = Task(name="download_release")
     db().add(task)
     db().commit()
     self._call_silently(task, tasks.DownloadReleaseTask, self.release_data)
     return task

    def message(cls, task, nodes_to_provisioning):
        logger.debug("ProvisionTask.message(task=%s)" % task.uuid)
        task = objects.Task.get_by_uid(
            task.id,
            fail_if_not_found=True,
            lock_for_update=True
        )
        objects.NodeCollection.lock_nodes(nodes_to_provisioning)
        serialized_cluster = provisioning_serializers.serialize(
            task.cluster, nodes_to_provisioning)

        for node in nodes_to_provisioning:
            if settings.FAKE_TASKS or settings.FAKE_TASKS_AMQP:
                continue
            logs_utils.prepare_syslog_dir(node)

        rpc_message = make_astute_message(
            task,
            cls._get_provision_method(task.cluster),
            'provision_resp',
            {
                'provisioning_info': serialized_cluster
            }
        )
        db().commit()
        return rpc_message

    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 _make_zabbix_request(cls, url, method, params, auth=None):
        header = {'Content-type': 'application/json'}
        data = {'jsonrpc': '2.0',
                'id': '1',
                'method': method,
                'params': params}
        if auth:
            data['auth'] = auth

        logger.debug("Zabbix request: %s", data)
        request = urllib2.Request(url, json.dumps(data), header)

        try:
            response = urllib2.urlopen(request)
        except urllib2.URLError as e:
            raise errors.CannotMakeZabbixRequest(
                "Can't make a request to Zabbix: {0}".format(e)
            )

        result = json.loads(response.read())
        logger.debug("Zabbix response: %s", result)

        if 'error' in result:
            code = result['error']['code']
            msg = result['error']['message']
            data = result['error'].get('data', '')
            raise errors.ZabbixRequestError(
                "Zabbix returned error code {0}, {1}: {2}".format(
                    code, msg, data
                )
            )

        return result['result']

 def make_bootable(self):
     logger.debug("Allocating /boot partition")
     self.create_partition(
         "/boot",
         self.vm.field_generator("calc_boot_size")
     )
     self.create_mbr(True)

 def create_mbr(self, boot=False):
     if self.free_space >= self.vm.field_generator("calc_mbr_size"):
         if boot:
             self.volumes.append({"type": "mbr"})
         logger.debug("Allocating MBR")
         self.free_space = self.free_space - \
             self.vm.field_generator("calc_mbr_size")

 def _traverse(self, cdict):
     new_dict = {}
     if isinstance(cdict, dict):
         for i, val in cdict.iteritems():
             if type(val) in (str, unicode, int, float):
                 new_dict[i] = val
             elif isinstance(val, dict):
                 if "generator" in val:
                     logger.debug("Generating value: generator: %s "
                                  "generator_args: %s", val["generator"],
                                  val.get("generator_args", []))
                     genval = self.field_generator(
                         val["generator"],
                         *(val.get("generator_args", []))
                     )
                     logger.debug("Generated value: %s", str(genval))
                     new_dict[i] = genval
                 else:
                     new_dict[i] = self._traverse(val)
             elif isinstance(val, list):
                 new_dict[i] = []
                 for d in val:
                     new_dict[i].append(self._traverse(d))
     elif isinstance(cdict, list):
         new_dict = []
         for d in cdict:
             new_dict.append(self._traverse(d))
     return new_dict

 def execute(cls, task):
     logger.debug("ProvisionTask.execute(task=%s)" % task.uuid)
     message = cls.message(task)
     task.cache = message
     db().add(task)
     db().commit()
     rpc.cast('naily', message)

 def execute(cls, task):
     logger.debug("DeploymentTask.execute(task=%s)" % task.uuid)
     message = cls.message(task)
     task.cache = message
     orm().add(task)
     orm().commit()
     rpc.cast('naily', message)

    def message(cls, task):
        logger.debug("DeploymentTask.message(task=%s)" % task.uuid)

        task.cluster.prepare_for_deployment()
        nodes = TaskHelper.nodes_to_deploy(task.cluster)
        nodes_ids = [n.id for n in nodes]
        for n in db().query(Node).filter_by(
                cluster=task.cluster).order_by(Node.id):
            # However, we must not pass nodes which are set to be deleted.
            if n.pending_deletion:
                continue

            if n.id in nodes_ids:  # It's node which we need to redeploy
                n.pending_addition = False
                if n.pending_roles:
                    n.roles += n.pending_roles
                    n.pending_roles = []
                if n.status in ('deploying'):
                    n.status = 'provisioned'
                n.progress = 0
                db().add(n)
                db().commit()

        # here we replace provisioning data if user redefined them
        serialized_cluster = task.cluster.replaced_deployment_info or \
            deployment_serializers.serialize(task.cluster)

        return {
            'method': 'deploy',
            'respond_to': 'deploy_resp',
            'args': {
                'task_uuid': task.uuid,
                'deployment_info': serialized_cluster}}

    def checkout(self, instance):
        fetch_file = os.path.join(
            const.REPOS_DIR,
            instance.repo_name,
            '.git/FETCH_HEAD'
        )
        if os.path.exists(fetch_file):
            current_ts = time.time()
            cluster = Cluster.get_by_uid(instance.env_id)
            last_fetch = os.stat(fetch_file).st_mtime
            if cluster.status != CLUSTER_STATUSES.deployment and \
                current_ts - last_fetch < const.REPO_TTL:
                return

        logger.debug("Repo TTL exceeded. Fetching code...")
        ssh_cmd = self._get_ssh_cmd(instance.repo_name)

        if not os.path.exists(self._get_key_path(instance.repo_name)):
            logger.debug('Key file does not exist. Creating...')
            self._create_key_file(instance.repo_name)

        with instance.repo.git.custom_environment(GIT_SSH=ssh_cmd):
            commit = instance.repo.remotes.origin.fetch(refspec=instance.ref)
            commit = commit[0].commit
            instance.repo.head.reference = commit
            instance.repo.head.reset(index=True, working_tree=True)

    def message(cls, task, nodes):
        logger.debug("%s.message(task=%s)", cls.__class__.__name__, task.uuid)

        for n in nodes:
            if n.pending_roles:
                n.roles += n.pending_roles
                n.pending_roles = []
            n.status = 'provisioned'
            n.progress = 0

        orchestrator_graph = deployment_graph.AstuteGraph(task.cluster)

        serialized_cluster = deployment_serializers.serialize(
            orchestrator_graph, task.cluster, nodes)

        # After serialization set pending_addition to False
        for node in nodes:
            node.pending_addition = False

        rpc_message = make_astute_message(
            task,
            'deploy',
            'deploy_resp',
            {
                'deployment_info': serialized_cluster
            }
        )
        db().flush()
        return rpc_message

    def update_interfaces_info(cls, node, update_by_agent=False):
        """Update interfaces in case of correct interfaces
        in meta field in node's model
        """
        try:
            cls.check_interfaces_correctness(node)
        except errors.InvalidInterfacesInfo as e:
            logger.debug("Cannot update interfaces: %s", e.message)
            return
        pxe_iface_name = cls._get_pxe_iface_name(node)
        for interface in node.meta["interfaces"]:
            # set 'pxe' property for appropriate iface
            if interface['name'] == pxe_iface_name:
                interface['pxe'] = True
            # try to get interface by mac address
            interface_db = next((
                n for n in node.nic_interfaces
                if utils.is_same_mac(n.mac, interface['mac'])),
                None)

            # try to get interface instance by interface name. this protects
            # us from loosing nodes when some NICs was replaced with a new one
            interface_db = interface_db or next((
                n for n in node.nic_interfaces if n.name == interface['name']),
                None)

            if interface_db:
                cls.__update_existing_interface(interface_db.id, interface,
                                                update_by_agent)
            else:
                cls.__add_new_interface(node, interface)

        cls.__delete_not_found_interfaces(node, node.meta["interfaces"])