On Tue, Apr 24, 2012 at 01:43:19PM +0300, Gary Kotton wrote: > On 04/24/2012 01:25 PM, Isaku Yamahata wrote: >> To run unit test, run run_tests.sh in quantum directory. >> In order to run unit tests of specified plugin, >> PLUGIN_DIR=<plugin dir> ./run_tests.sh [-V] >> >> For example >> PLUGIN_DIR=quantum/plugins/ryu ./run_tests.sh [-V] > Thanks >> >> On Tue, Apr 24, 2012 at 09:57:27AM +0300, Gary Kotton wrote: >>> Hi, >>> Can you please look at the attached file. Any comments would be greatly >>> appreciated. How can I run the unit tests on this code? >> Are you seeing transaction abort? Interesting. > Yes. >> I have never seen the abort. I haven't heavily test it so far, though. >> Just from curiosity, how many compute nodes are you running? > At the moment I am running 1 node. I encountered the problem when I > rebooted the node that was running the quantum server and plugin. I > originally encountered the problem with the ovs plugin. I validated that > it took place with the bridge linuxplugin too. > >> If transaction abort is the real issue, it is the common issue >> between ovs, linuxbridge and ryu plugins. > Correct. I just looked at the agents and not the plugins.
Okay, then let's file a bug report and discuss at openstack ml. thanks, >> Although there would be several approaches I can think of, >> (your approach making agent abort-tolerable, table locking, marking deleted >> instead of removing column), >> it should be discussed as quantum issue, not Ryu specific issue. >> >> thanks, >> >>> Thanks >>> Gary >>> >>> On 04/24/2012 09:02 AM, Isaku Yamahata wrote: >>>> Cc to ryu-devel. >>>> >>>> On Tue, Apr 24, 2012 at 05:47:43AM -0000, Gary Kotton wrote: >>>>> Hi, >>>>> I found a number of problems with the Quantum agents. These happen when >>>>> the Quantum server is down (for example the host running the server is >>>>> rebooted). >>>>> I would like to update the RYU agent code to deal with these problems. >>>>> Are there uint tests that I can run to check this? >>>>> Thanks >>>> Thank you for testing! I'm afraid Ryu plugin has poor unit test for now. >>>> It's the area that we have to address. (patches are very welcome, off >>>> course) >>>> >>>> Does the attached patch help? >>>> Although it might not apply cleanly because it's for the gre demo tree, >>>> you get the idea. >>>> If you're seeing different issue, can you please elaborate on it? >>>> >>>> thanks, >>>> >>>> commit a043d85c63d93d6812937c034640ebe892670c05 >>>> Author: Isaku Yamahata<[email protected]> >>>> Date: Tue Apr 10 11:56:07 2012 +0900 >>>> >>>> plugins/ryu/agent: fix exception >>>> >>>> When starting up, the network is expected to be created by quantum >>>> server. >>>> However there is a race between quantum server startup and quantum >>>> agent >>>> startup. So have agent also make the network. >>>> >>>> Traceback (most recent call last): >>>> File >>>> "/opt/stack/quantum/quantum/plugins/ryu/agent/ryu_quantum_agent.py", line >>>> 540, in<module> >>>> main() >>>> File >>>> "/opt/stack/quantum/quantum/plugins/ryu/agent/ryu_quantum_agent.py", line >>>> 534, in main >>>> plugin.daemon_loop() >>>> File >>>> "/opt/stack/quantum/quantum/plugins/ryu/agent/ryu_quantum_agent.py", line >>>> 490, in daemon_loop >>>> self.gre_ports.update() >>>> File >>>> "/opt/stack/quantum/quantum/plugins/ryu/agent/ryu_quantum_agent.py", line >>>> 367, in update >>>> self._add_port(node) >>>> File >>>> "/opt/stack/quantum/quantum/plugins/ryu/agent/ryu_quantum_agent.py", line >>>> 344, in _add_port >>>> self.int_br.datapath_id, ofport) >>>> File "/opt/stack/ryu/ryu/app/client.py", line 91, in create_port >>>> self._do_request('POST', self.path_port % (network_id, dpid, >>>> port)) >>>> File "/opt/stack/ryu/ryu/app/client.py", line 55, in _do_request >>>> res.getheaders(), res.read()) >>>> httplib.HTTPException: (<httplib.HTTPResponse instance at >>>> 0x25e7a28>, 'code 404 reason Resource not found', [('date', 'Tue, 10 Apr >>>> 2012 02:26:03 GMT'), ('content-length', '466'), ('content-type', >>>> 'application/json')], '{"displayError": "The server does not have data for >>>> the request.", "error": "You submitted the following request.\\n\\n >>>> POST /v1.0/networks/__NW_ID_VPORT_GRE__/00004e7f48e36243_\\n\\nThis >>>> request is not valid. The following paths were evaluated and failed\\nfor >>>> the indicated reason.\\n\\n - >>>> /networks/{network-id}/{dpid}_{port-id}\\n Invalid format: >>>> 00004e7f48e36243_\\n\\nYou can get a list of all valid requests with the >>>> following request.\\n\\n GET /v1.0/doc"}') >>>> >>>> Signed-off-by: Isaku Yamahata<[email protected]> >>>> >>>> diff --git a/quantum/plugins/ryu/agent/ryu_quantum_agent.py >>>> b/quantum/plugins/ryu/agent/ryu_quantum_agent.py >>>> index a46ff2b..78b2d9e 100755 >>>> --- a/quantum/plugins/ryu/agent/ryu_quantum_agent.py >>>> +++ b/quantum/plugins/ryu/agent/ryu_quantum_agent.py >>>> @@ -321,6 +321,7 @@ class GREPortSet(object): >>>> def setup(self): >>>> _ovs_node_update(self.db, self.int_br.datapath_id, >>>> self.tunnel_ip) >>>> >>>> + self.api.update_network(rest_nw_id.NW_ID_VPORT_GRE) >>>> for port in self.int_br.get_gre_ports(): >>>> try: >>>> node = self.db.ovs_node.filter( >>>> @@ -385,6 +386,7 @@ class VifPortSet(object): >>>> self.api.update_port(network_id, port.switch.datapath_id, >>>> port.ofport) >>>> if port.vif_mac is not None: >>>> # external port doesn't have mac address >>>> + self.api.update_network(network_id) >>>> self.api.update_mac(network_id, port.switch.datapath_id, >>>> port.ofport, port.vif_mac) >>>> else: >>>> >>>> >>> #!/usr/bin/env python >>> # vim: tabstop=4 shiftwidth=4 softtabstop=4 >>> # >>> # Copyright 2012 Isaku Yamahata<yamahata at private email ne jp> >>> # Based on openvswitch agent. >>> # >>> # Copyright 2011 Nicira Networks, Inc. >>> # 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. >>> # @author: Isaku Yamahata >>> import ConfigParser >>> import logging as LOG >>> import shlex >>> import signal >>> import sys >>> import time >>> from optparse import OptionParser >>> from sqlalchemy.ext.sqlsoup import SqlSoup >>> from subprocess import PIPE, Popen >>> >>> from ryu.app import rest_nw_id >>> from ryu.app.client import OFPClient >>> >>> >>> OP_STATUS_UP = "UP" >>> OP_STATUS_DOWN = "DOWN" >>> # Database connctivity status >>> DB_OK = 1 >>> DB_RECONNECT = 2 >>> >>> >>> class VifPort: >>> """ >>> A class to represent a VIF (i.e., a port that has 'iface-id' and >>> 'vif-mac' >>> attributes set). >>> """ >>> def __init__(self, port_name, ofport, vif_id, vif_mac, switch): >>> self.port_name = port_name >>> self.ofport = ofport >>> self.vif_id = vif_id >>> self.vif_mac = vif_mac >>> self.switch = switch >>> >>> def __str__(self): >>> return ("iface-id=%s, vif_mac=%s, port_name=%s, ofport=%s, " >>> "bridge name = %s" % (self.vif_id, >>> self.vif_mac, >>> self.port_name, >>> self.ofport, >>> self.switch.br_name)) >>> >>> >>> class OVSBridge: >>> def __init__(self, br_name, root_helper): >>> self.br_name = br_name >>> self.root_helper = root_helper >>> self.datapath_id = None >>> >>> def find_datapath_id(self): >>> # ovs-vsctl get Bridge br-int datapath_id >>> res = self.run_vsctl(["get", "Bridge", self.br_name, >>> "datapath_id"]) >>> >>> # remove preceding/trailing double quotes >>> dp_id = res.strip().strip('"') >>> self.datapath_id = dp_id >>> >>> def run_cmd(self, args): >>> cmd = shlex.split(self.root_helper) + args >>> pipe = Popen(cmd, stdout=PIPE) >>> retval = pipe.communicate()[0] >>> if pipe.returncode == -(signal.SIGALRM): >>> LOG.debug("## timeout running command: " + " ".join(cmd)) >>> return retval >>> >>> def run_vsctl(self, args): >>> full_args = ["ovs-vsctl", "--timeout=2"] + args >>> return self.run_cmd(full_args) >>> >>> def set_controller(self, target): >>> methods = ("ssl", "tcp", "unix", "pssl", "ptcp", "punix") >>> args = target.split(":") >>> if not args[0] in methods: >>> target = "tcp:" + target >>> self.run_vsctl(["set-controller", self.br_name, target]) >>> >>> def db_get_map(self, table, record, column): >>> str_ = self.run_vsctl(["get", table, record, >>> column]).rstrip("\n\r") >>> return self.db_str_to_map(str_) >>> >>> def db_get_val(self, table, record, column): >>> return self.run_vsctl(["get", table, record, >>> column]).rstrip("\n\r") >>> >>> @staticmethod >>> def db_str_to_map(full_str): >>> list = full_str.strip("{}").split(", ") >>> ret = {} >>> for elem in list: >>> if elem.find("=") == -1: >>> continue >>> arr = elem.split("=") >>> ret[arr[0]] = arr[1].strip("\"") >>> return ret >>> >>> def get_port_name_list(self): >>> res = self.run_vsctl(["list-ports", self.br_name]) >>> return res.split("\n")[:-1] >>> >>> def get_xapi_iface_id(self, xs_vif_uuid): >>> return self.run_cmd( >>> ["xe", >>> "vif-param-get", >>> "param-name=other-config", >>> "param-key=nicira-iface-id", >>> "uuid=%s" % xs_vif_uuid]).strip() >>> >>> def _vifport(self, name, external_ids): >>> ofport = self.db_get_val("Interface", name, "ofport") >>> return VifPort(name, ofport, external_ids["iface-id"], >>> external_ids["attached-mac"], self) >>> >>> def _get_ports(self, get_port): >>> ports = [] >>> port_names = self.get_port_name_list() >>> for name in port_names: >>> port = get_port(name) >>> if port: >>> ports.append(port) >>> >>> return ports >>> >>> def _get_vif_port(self, name): >>> external_ids = self.db_get_map("Interface", name, "external_ids") >>> if "iface-id" in external_ids and "attached-mac" in external_ids: >>> return self._vifport(name, external_ids) >>> elif ("xs-vif-uuid" in external_ids and >>> "attached-mac" in external_ids): >>> # if this is a xenserver and iface-id is not automatically >>> # synced to OVS from XAPI, we grab it from XAPI directly >>> ofport = self.db_get_val("Interface", name, "ofport") >>> iface_id = self.get_xapi_iface_id(external_ids["xs-vif-uuid"]) >>> return VifPort(name, ofport, iface_id, >>> external_ids["attached-mac"], self) >>> >>> def get_vif_ports(self): >>> "returns a VIF object for each VIF port" >>> return self._get_ports(self._get_vif_port) >>> >>> def _get_external_port(self, name): >>> external_ids = self.db_get_map("Interface", name, "external_ids") >>> if external_ids: >>> return >>> >>> ofport = self.db_get_val("Interface", name, "ofport") >>> return VifPort(name, ofport, None, None, self) >>> >>> def get_external_ports(self): >>> return self._get_ports(self._get_external_port) >>> >>> >>> def check_ofp_mode(db): >>> LOG.debug("checking db") >>> >>> ofp_controller_addr = None >>> ofp_rest_api_addr = None >>> >>> try: >>> servers = db.ofp_server.all() >>> except: >>> self.db_state = DB_RECONNECT >>> return (ofp_controller_addr, ofp_rest_api_addr) >>> >>> ofp_controller_addr = None >>> ofp_rest_api_addr = None >>> for serv in servers: >>> if serv.host_type == "REST_API": >>> ofp_rest_api_addr = serv.address >>> elif serv.host_type == "controller": >>> ofp_controller_addr = serv.address >>> else: >>> LOG.warn("ignoring unknown server type %s", serv) >>> >>> LOG.debug("controller %s", ofp_controller_addr) >>> LOG.debug("api %s", ofp_rest_api_addr) >>> if not ofp_controller_addr: >>> raise RuntimeError("OF controller isn't specified") >>> if not ofp_rest_api_addr: >>> raise RuntimeError("Ryu rest API port isn't specified") >>> >>> LOG.debug("going to ofp controller mode %s %s", >>> ofp_controller_addr, ofp_rest_api_addr) >>> return (ofp_controller_addr, ofp_rest_api_addr) >>> >>> >>> class OVSQuantumOFPRyuAgent: >>> def __init__(self, integ_br, root_helper): >>> self.root_helper = root_helper >>> self. >>> (ofp_controller_addr, ofp_rest_api_addr) = check_ofp_mode(db) >>> >>> self.nw_id_external = rest_nw_id.NW_ID_EXTERNAL >>> self.api = OFPClient(ofp_rest_api_addr) >>> self._setup_integration_br(integ_br, ofp_controller_addr) >>> >>> def _setup_integration_br(self, integ_br, ofp_controller_addr): >>> self.int_br = OVSBridge(integ_br, self.root_helper) >>> self.int_br.find_datapath_id() >>> self.int_br.set_controller(ofp_controller_addr) >>> for port in self.int_br.get_external_ports(): >>> self._port_update(self.nw_id_external, port) >>> >>> def _port_update(self, network_id, port): >>> self.api.update_port(network_id, port.switch.datapath_id, >>> port.ofport) >>> >>> def _all_bindings(self, db): >>> """return interface id -> port which include network id >>> bindings""" >>> try: >>> return dict((port.interface_id, port) for port in >>> db.ports.all()) >>> except: >>> self.db_state = DB_RECONNECT >>> return {} >>> >>> def daemon_loop(self, options): >>> count = 0 >>> threshold = 1 >>> >>> db = SqlSoup(options["sql_connection"]) >>> LOG.info("Connecting to database \"%s\" on %s", >>> db.engine.url.database, db.engine.url.host) >>> >>> (ofp_controller_addr, ofp_rest_api_addr) = check_ofp_mode(db) >>> >>> self.nw_id_external = rest_nw_id.NW_ID_EXTERNAL >>> self.api = OFPClient(ofp_rest_api_addr) >>> self._setup_integration_br(integ_br, ofp_controller_addr) >>> >>> # on startup, register all existing ports >>> all_bindings = self._all_bindings(db) >>> >>> local_bindings = {} >>> vif_ports = {} >>> for port in self.int_br.get_vif_ports(): >>> vif_ports[port.vif_id] = port >>> if port.vif_id in all_bindings: >>> net_id = all_bindings[port.vif_id].network_id >>> local_bindings[port.vif_id] = net_id >>> self._port_update(net_id, port) >>> all_bindings[port.vif_id].op_status = OP_STATUS_UP >>> LOG.info("Updating binding to net-id = %s for %s", >>> net_id, str(port)) >>> try: >>> db.commit() >>> except: >>> db.rollback() >>> >>> old_vif_ports = vif_ports >>> old_local_bindings = local_bindings >>> >>> while True: >>> if self.db_state == DB_RECONNECT: >>> time.sleep(2) >>> count = count + 1 >>> if count>= threshold: >>> LOG.info("Problem connecting to database \"%s\" on %s" >>> (db.engine.url.database, db.engine.url.host)) >>> count = 0 >>> threshold = threshold * 10 >>> db = SqlSoup(options["sql_connection"]) >>> self.db_state = DB_OK >>> LOG.info("Re-connecting to database \"%s\" on %s" % >>> (db.engine.url.database, db.engine.url.host)) >>> (ofp_controller_addr, ofp_rest_api_addr) = >>> check_ofp_mode(db) >>> if self.db_state == DB_RECONNECT: >>> continue >>> self.api = OFPClient(ofp_rest_api_addr) >>> >>> all_bindings = self._all_bindings(db) >>> if self.db_state == DB_RECONNECT: >>> continue >>> >>> new_vif_ports = {} >>> new_local_bindings = {} >>> for port in self.int_br.get_vif_ports(): >>> new_vif_ports[port.vif_id] = port >>> if port.vif_id in all_bindings: >>> net_id = all_bindings[port.vif_id].network_id >>> new_local_bindings[port.vif_id] = net_id >>> >>> old_b = old_local_bindings.get(port.vif_id) >>> new_b = new_local_bindings.get(port.vif_id) >>> if old_b == new_b: >>> continue >>> >>> if not old_b: >>> LOG.info("Removing binding to net-id = %s for %s", >>> old_b, str(port)) >>> if port.vif_id in all_bindings: >>> all_bindings[port.vif_id].op_status = >>> OP_STATUS_DOWN >>> if not new_b: >>> if port.vif_id in all_bindings: >>> all_bindings[port.vif_id].op_status = OP_STATUS_UP >>> LOG.info("Adding binding to net-id = %s for %s", >>> new_b, str(port)) >>> >>> for vif_id in old_vif_ports: >>> if vif_id not in new_vif_ports: >>> LOG.info("Port Disappeared: %s", vif_id) >>> if vif_id in all_bindings: >>> all_bindings[vif_id].op_status = OP_STATUS_DOWN >>> >>> old_vif_ports = new_vif_ports >>> old_local_bindings = new_local_bindings >>> try: >>> db.commit() >>> except: >>> db.rollback() >>> >>> time.sleep(2) >>> >>> >>> def main(): >>> usagestr = "%prog [OPTIONS]<config file>" >>> parser = OptionParser(usage=usagestr) >>> parser.add_option("-v", "--verbose", dest="verbose", >>> action="store_true", default=False, help="turn on verbose logging") >>> >>> options, args = parser.parse_args() >>> >>> if options.verbose: >>> LOG.basicConfig(level=LOG.DEBUG) >>> else: >>> LOG.basicConfig(level=LOG.WARN) >>> >>> if len(args) != 1: >>> parser.print_help() >>> sys.exit(1) >>> >>> config_file = args[0] >>> config = ConfigParser.ConfigParser() >>> try: >>> config.read(config_file) >>> except Exception, e: >>> LOG.error("Unable to parse config file \"%s\": %s", >>> config_file, str(e)) >>> >>> integ_br = config.get("OVS", "integration-bridge") >>> >>> root_helper = config.get("AGENT", "root_helper") >>> >>> options = {"sql_connection": config.get("DATABASE", "sql_connection")} >>> db = SqlSoup(options["sql_connection"]) >>> >>> LOG.info("Connecting to database \"%s\" on %s", >>> db.engine.url.database, db.engine.url.host) >>> plugin = OVSQuantumOFPRyuAgent(integ_br, root_helper) >>> plugin.daemon_loop(options) >>> >>> sys.exit(0) >>> >>> >>> if __name__ == "__main__": >>> main() >> > -- yamahata ------------------------------------------------------------------------------ Live Security Virtual Conference Exclusive live event will cover all the ways today's security and threat landscape has changed and how IT managers can respond. Discussions will include endpoint security, mobile security and the latest in malware threats. http://www.accelacomm.com/jaw/sfrnl04242012/114/50122263/ _______________________________________________ Ryu-devel mailing list [email protected] https://lists.sourceforge.net/lists/listinfo/ryu-devel
