Implement Zebra protocol client service

This patch implements APIs for performing as a protocol daemon of Quagga
such as "ripd", "ospfd", "bgpd", for example.

Signed-off-by: IWASE Yusuke <iwase.yusuke0@gmail.com>
Signed-off-by: FUJITA Tomonori <fujita.tomonori@lab.ntt.co.jp>

7 files changed, 584 insertions, 1 deletions

diff --git a/ryu/flags.py b/ryu/flags.py
index 43403b5c..11ce83d0 100644
--- a/ryu/flags.py
+++ b/ryu/flags.py
@@ -62,7 +62,6 @@ CONF.register_cli_opts([
 DEFAULT_RPC_PORT = 50002
 DEFAULT_RPC_HOST = '0.0.0.0'
 
-CONF = cfg.CONF
 CONF.register_cli_opts([
     cfg.IntOpt('rpc-port', default=DEFAULT_RPC_PORT,
                help='Port for RPC server (default: %s)' % DEFAULT_RPC_PORT),
@@ -72,3 +71,34 @@ CONF.register_cli_opts([
                help='The config file formatted in Python source file. '
                     'Please refer to "bgp_sample_conf.py" for details.')
 ], group='bgp-app')
+
+
+DEFAULT_ZSERV_HOST = '/var/run/quagga/zserv.api'
+DEFAULT_ZSERV_PORT = 2600
+DEFAULT_ZSERV_VERSION = 2  # Version of Ubuntu 16.04 LTS packaged Quagga
+DEFAULT_ZSERV_CLIENT_ROUTE_TYPE = 'BGP'
+DEFAULT_ZSERV_INTERVAL = 10
+
+CONF.register_cli_opts([
+    cfg.StrOpt(
+        'server-host', default=DEFAULT_ZSERV_HOST,
+        help='Path to Unix Socket or IP address of Zebra server '
+             '(default: %s)' % DEFAULT_ZSERV_HOST),
+    cfg.IntOpt(
+        'server-port', default=DEFAULT_ZSERV_PORT,
+        help='Port number of Zebra server '
+             '(default: %s)'
+        % DEFAULT_ZSERV_PORT),
+    cfg.IntOpt(
+        'server-version', default=DEFAULT_ZSERV_VERSION,
+        help='Zebra protocol version of Zebra server '
+             '(default: %s)' % DEFAULT_ZSERV_VERSION),
+    cfg.StrOpt(
+        'client-route-type', default=DEFAULT_ZSERV_CLIENT_ROUTE_TYPE,
+        help='Zebra route type advertised by Zebra client service. '
+             '(default: %s)' % DEFAULT_ZSERV_CLIENT_ROUTE_TYPE),
+    cfg.IntOpt(
+        'retry-interval', default=DEFAULT_ZSERV_INTERVAL,
+        help='Retry interval connecting to Zebra server '
+             '(default: %s)' % DEFAULT_ZSERV_INTERVAL),
+], group='zapi')
diff --git a/ryu/lib/packet/zebra.py b/ryu/lib/packet/zebra.py
index 1e7006e1..fec37b42 100644
--- a/ryu/lib/packet/zebra.py
+++ b/ryu/lib/packet/zebra.py
@@ -1250,6 +1250,8 @@ class _ZebraIPRoute(_ZebraMessageBody):
         prefix = _serialize_ip_prefix(self.prefix)
 
         nexthops = _serialize_nexthops(self.nexthops)
+        if self.nexthops:
+            self.message |= ZAPI_MESSAGE_NEXTHOP  # fixup
 
         options = self._serialize_message_option(
             self.distance, ZAPI_MESSAGE_DISTANCE, '!B')
diff --git a/ryu/services/protocols/zebra/__init__.py b/ryu/services/protocols/zebra/__init__.py
new file mode 100644
index 00000000..3f19486e
--- /dev/null
+++ b/ryu/services/protocols/zebra/__init__.py
@@ -0,0 +1,18 @@
+# Copyright (C) 2017 Nippon Telegraph and Telephone Corporation.
+#
+# 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.
+
+"""
+Implementation for Zebra protocol service.
+"""
diff --git a/ryu/services/protocols/zebra/client/__init__.py b/ryu/services/protocols/zebra/client/__init__.py
new file mode 100644
index 00000000..eadba0af
--- /dev/null
+++ b/ryu/services/protocols/zebra/client/__init__.py
@@ -0,0 +1,20 @@
+# Copyright (C) 2017 Nippon Telegraph and Telephone Corporation.
+#
+# 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.
+
+"""
+Client implementation for Zebra protocol service.
+
+This module provides the client side implementation for Zebra protocol.
+"""
diff --git a/ryu/services/protocols/zebra/client/event.py b/ryu/services/protocols/zebra/client/event.py
new file mode 100644
index 00000000..4995c4df
--- /dev/null
+++ b/ryu/services/protocols/zebra/client/event.py
@@ -0,0 +1,46 @@
+# Copyright (C) 2017 Nippon Telegraph and Telephone Corporation.
+#
+# 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.
+
+"""
+Events generated by Zebra Client service.
+"""
+
+from ryu.controller.event import EventBase
+
+
+class EventZClientBase(EventBase):
+    """
+    The base class for the event generated by ZClient.
+    """
+
+
+class EventZServConnected(EventZClientBase):
+    """
+    The event class for notifying the connection to Zebra server.
+    """
+
+    def __init__(self, zserv):
+        super(EventZServConnected, self).__init__()
+        self.zserv = zserv
+
+
+class EventZServDisconnected(EventZClientBase):
+    """
+    The event class for notifying the disconnection from Zebra server.
+    """
+
+    def __init__(self, zserv):
+        super(EventZServDisconnected, self).__init__()
+        self.zserv = zserv
diff --git a/ryu/services/protocols/zebra/client/zclient.py b/ryu/services/protocols/zebra/client/zclient.py
new file mode 100644
index 00000000..59d659d3
--- /dev/null
+++ b/ryu/services/protocols/zebra/client/zclient.py
@@ -0,0 +1,345 @@
+# Copyright (C) 2017 Nippon Telegraph and Telephone Corporation.
+#
+# 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.
+
+"""
+Zebra Client corresponding to 'zclient' structure.
+"""
+
+import os
+import socket
+import struct
+
+import netaddr
+
+from ryu import cfg
+from ryu.base.app_manager import RyuApp
+from ryu.lib import hub
+from ryu.lib import ip
+from ryu.lib.packet import zebra
+from ryu.lib.packet import safi as packet_safi
+from ryu.services.protocols.zebra import event
+from ryu.services.protocols.zebra.client import event as zclient_event
+
+
+CONF = cfg.CONF['zapi']
+GLOBAL_CONF = cfg.CONF
+
+
+def create_connection(address):
+    """
+    Wrapper for socket.create_connection() function.
+
+    If *address* (a 2-tuple ``(host, port)``) contains a valid IPv4/v6
+    address, passes *address* to socket.create_connection().
+    If *host* is valid path to Unix Domain socket, tries to connect to
+    the server listening on the given socket.
+
+    :param address: IP address or path to Unix Domain socket.
+    :return: Socket instance.
+    """
+    host, _port = address
+
+    if (netaddr.valid_ipv4(host)
+            or netaddr.valid_ipv6(host)):
+        return socket.create_connection(address)
+    elif os.path.exists(host):
+        sock = None
+        try:
+            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+            sock.connect(host)
+        except socket.error as e:
+            if sock is not None:
+                sock.close()
+            raise e
+        return sock
+    else:
+        raise ValueError('Invalid IP address or Unix Socket: %s' % host)
+
+
+def get_zebra_route_type_by_name(route_type='BGP'):
+    """
+    Returns the constant value for Zebra route type named "ZEBRA_ROUTE_*"
+    from its name.
+
+    See "ZEBRA_ROUTE_*" constants in "ryu.lib.packet.zebra" module.
+
+    :param route_type: Route type name (e.g., Kernel, BGP).
+    :return: Constant value for Zebra route type.
+    """
+    return getattr(zebra, "ZEBRA_ROUTE_%s" % route_type.upper())
+
+
+class ZServer(object):
+    """
+    Zebra server class.
+    """
+
+    def __init__(self, client):
+        self.client = client
+        self.logger = client.logger
+        self.is_active = False
+        self.sock = None  # Client socket connecting to Zebra server
+        self.threads = []
+
+    def start(self):
+        self.is_active = True
+        try:
+            self.sock = create_connection(self.client.zserv_addr)
+        except socket.error as e:
+            self.logger.exception(
+                'Cannot connect to Zebra server%s: %s',
+                self.client.zserv_addr, e)
+            self.stop()
+            return None
+
+        self.sock.settimeout(GLOBAL_CONF.socket_timeout)
+
+        self.threads.append(hub.spawn(self._send_loop))
+        self.threads.append(hub.spawn(self._recv_loop))
+
+        # Send the following messages at starting connection.
+        # - ZEBRA_HELLO to register route_type
+        # - ZEBRA_ROUTER_ID_ADD to get router_id
+        # - ZEBRA_INTERFACE_ADD to get info for interfaces
+        self.client.send_msg(
+            zebra.ZebraMessage(
+                version=self.client.zserv_ver,
+                body=zebra.ZebraHello(self.client.route_type)))
+        self.client.send_msg(
+            zebra.ZebraMessage(
+                version=self.client.zserv_ver,
+                command=zebra.ZEBRA_ROUTER_ID_ADD))
+        self.client.send_msg(
+            zebra.ZebraMessage(
+                version=self.client.zserv_ver,
+                command=zebra.ZEBRA_INTERFACE_ADD))
+
+        self.client.send_event_to_observers(
+            zclient_event.EventZServConnected(self))
+
+        hub.joinall(self.threads)
+
+        self.client.send_event_to_observers(
+            zclient_event.EventZServDisconnected(self))
+
+    def stop(self):
+        self.is_active = False
+
+    def _send_loop(self):
+        try:
+            while self.is_active:
+                buf = self.client.send_q.get()
+                self.sock.sendall(buf)
+        except socket.error as e:
+            self.logger.exception(
+                'Error while sending message to Zebra server%s: %s',
+                self.client.zserv_addr, e)
+
+        self.stop()
+
+    def _recv_loop(self):
+        buf = b''
+        min_len = recv_len = zebra.ZebraMessage.get_header_size(
+            self.client.zserv_ver)
+        try:
+            while self.is_active:
+                try:
+                    recv_buf = self.sock.recv(recv_len)
+                except socket.timeout:
+                    continue
+
+                if len(recv_buf) == 0:
+                    break
+
+                buf += recv_buf
+                while len(buf) >= min_len:
+                    (length,) = struct.unpack_from('!H', buf)
+                    if (length - len(buf)) > 0:
+                        # Need to receive remaining data
+                        recv_len = length - len(buf)
+                        break
+
+                    msg, _, buf = zebra.ZebraMessage.parser(buf)
+
+                    ev = event.message_to_event(self.client, msg)
+                    if ev:
+                        self.client.send_event_to_observers(ev)
+
+        except socket.error as e:
+            self.logger.exception(
+                'Error while sending message to Zebra server%s: %s',
+                self.client.zserv_addr, e)
+
+        self.stop()
+
+
+class ZClient(RyuApp):
+    """
+    The base class for Zebra client application.
+    """
+    _EVENTS = event.ZEBRA_EVENTS + [
+        zclient_event.EventZServConnected,
+        zclient_event.EventZServDisconnected,
+    ]
+
+    def __init__(self, *args, **kwargs):
+        super(ZClient, self).__init__(*args, **kwargs)
+        self.zserv = None  # ZServer instance
+        self.zserv_addr = (CONF.server_host, CONF.server_port)
+        self.zserv_ver = CONF.server_version
+        self.send_q = hub.Queue(16)
+        self.route_type = get_zebra_route_type_by_name(
+            CONF.client_route_type)
+
+    def start(self):
+        super(ZClient, self).start()
+
+        return hub.spawn(self._service_loop)
+
+    def _service_loop(self):
+        while self.is_active:
+            self.zserv = ZServer(self)
+            self.zserv.start()
+
+            hub.sleep(CONF.retry_interval)
+
+        self.close()
+
+    def close(self):
+        self.is_active = False
+        self._send_event(self._event_stop, None)
+        self.zserv.stop()
+
+    def send_msg(self, msg):
+        """
+        Sends Zebra message.
+
+        :param msg: Instance of py:class: `ryu.lib.packet.zebra.ZebraMessage`.
+        :return: Serialized msg if succeeded, otherwise None.
+        """
+        if not self.is_active:
+            self.logger.debug(
+                'Cannot send message: Already deactivated: msg=%s', msg)
+            return
+        elif not self.send_q:
+            self.logger.debug(
+                'Cannot send message: Send queue does not exist: msg=%s', msg)
+            return
+        elif self.zserv_ver != msg.version:
+            self.logger.debug(
+                'Zebra protocol version mismatch:'
+                'server_version=%d, msg.version=%d',
+                self.zserv_ver, msg.version)
+            msg.version = self.zserv_ver  # fixup
+
+        self.send_q.put(msg.serialize())
+
+    def _send_ip_route_impl(
+            self, prefix, nexthops=None,
+            safi=packet_safi.UNICAST, flags=zebra.ZEBRA_FLAG_INTERNAL,
+            distance=None, metric=None, mtu=None, tag=None,
+            is_withdraw=False):
+        if ip.valid_ipv4(prefix):
+            if is_withdraw:
+                msg_cls = zebra.ZebraIPv4RouteDelete
+            else:
+                msg_cls = zebra.ZebraIPv4RouteAdd
+        elif ip.valid_ipv6(prefix):
+            if is_withdraw:
+                msg_cls = zebra.ZebraIPv6RouteDelete
+            else:
+                msg_cls = zebra.ZebraIPv6RouteAdd
+        else:
+            raise ValueError('Invalid prefix: %s' % prefix)
+
+        nexthop_list = []
+        for nexthop in nexthops:
+            if netaddr.valid_ipv4(nexthop):
+                nexthop_list.append(zebra.NextHopIPv4(addr=nexthop))
+            elif netaddr.valid_ipv6(nexthop):
+                nexthop_list.append(zebra.NextHopIPv6(addr=nexthop))
+            else:
+                raise ValueError('Invalid nexthop: %s' % nexthop)
+
+        msg = zebra.ZebraMessage(
+            version=self.zserv_ver,
+            body=msg_cls(
+                route_type=self.route_type,
+                flags=flags,
+                message=0,
+                safi=safi,
+                prefix=prefix,
+                nexthops=nexthop_list,
+                distance=distance,
+                metric=metric,
+                mtu=mtu,
+                tag=tag))
+        self.send_msg(msg)
+
+        return msg
+
+    def send_ip_route_add(
+            self, prefix, nexthops=None,
+            safi=packet_safi.UNICAST, flags=zebra.ZEBRA_FLAG_INTERNAL,
+            distance=None, metric=None, mtu=None, tag=None):
+        """
+        Sends ZEBRA_IPV4/v6_ROUTE_ADD message to Zebra daemon.
+
+        :param prefix: IPv4/v6 Prefix to advertise.
+        :param nexthops: List of nexthop addresses.
+        :param safi: SAFI to advertise.
+        :param flags: Message flags to advertise. See "ZEBRA_FLAG_*".
+        :param distance: (Optional) Distance to advertise.
+        :param metric: (Optional) Metric to advertise.
+        :param mtu: (Optional) MTU size to advertise.
+        :param tag: (Optional) TAG information to advertise.
+        :return: Zebra message instance to be sent. None if failed.
+        """
+        try:
+            return self._send_ip_route_impl(
+                prefix=prefix, nexthops=nexthops, safi=safi, flags=flags,
+                distance=distance, metric=metric, mtu=mtu, tag=tag,
+                is_withdraw=False)
+        except ValueError as e:
+            self.logger.exception(
+                'Cannot send IP route add message: %s', e)
+            return None
+
+    def send_ip_route_delete(
+            self, prefix, nexthops=None,
+            safi=packet_safi.UNICAST, flags=zebra.ZEBRA_FLAG_INTERNAL,
+            distance=None, metric=None, mtu=None, tag=None):
+        """
+        Sends ZEBRA_IPV4/v6_ROUTE_DELETE message to Zebra daemon.
+
+        :param prefix: IPv4/v6 Prefix to advertise.
+        :param nexthops: List of nexthop addresses.
+        :param safi: SAFI to advertise.
+        :param flags: Message flags to advertise. See "ZEBRA_FLAG_*".
+        :param distance: (Optional) Distance to advertise.
+        :param metric: (Optional) Metric to advertise.
+        :param mtu: (Optional) MTU size to advertise.
+        :param tag: (Optional) TAG information to advertise.
+        :return: Zebra message instance to be sent. None if failed.
+        """
+        try:
+            return self._send_ip_route_impl(
+                prefix=prefix, nexthops=nexthops, safi=safi, flags=flags,
+                distance=distance, metric=metric, mtu=mtu, tag=tag,
+                is_withdraw=True)
+        except ValueError as e:
+            self.logger.exception(
+                'Cannot send IP route delete message: %s', e)
+            return None
diff --git a/ryu/services/protocols/zebra/event.py b/ryu/services/protocols/zebra/event.py
new file mode 100644
index 00000000..030909c3
--- /dev/null
+++ b/ryu/services/protocols/zebra/event.py
@@ -0,0 +1,122 @@
+# Copyright (C) 2017 Nippon Telegraph and Telephone Corporation.
+#
+# 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.
+
+"""
+Events for Zebra protocol service.
+"""
+
+import inspect
+import logging
+
+from ryu import utils
+from ryu.controller import event
+from ryu.lib.packet import zebra
+
+
+LOG = logging.getLogger(__name__)
+MOD = utils.import_module(__name__)
+
+ZEBRA_EVENTS = []
+
+
+class EventZebraBase(event.EventBase):
+    """
+    The base class for Zebra protocol service event class.
+
+    The subclasses have at least ``zclient`` and the same attributes with
+    :py:class: `ryu.lib.packet.zebra.ZebraMessage`.
+    ``zclient`` is an instance of Zebra client class. See
+    :py:class: `ryu.services.protocols.zebra.client.zclient.ZClient` or
+    :py:class: `ryu.services.protocols.zebra.server.zserver.ZClient`.
+
+    The subclasses are named as::
+
+        ``"Event" + <Zebra message body class name>``
+
+    For Example, if the service received ZEBRA_INTERFACE_ADD message,
+    the body class should be
+    :py:class: `ryu.lib.packet.zebra.ZebraInterfaceAdd`, then the event
+    class will be named as::
+
+        "Event" + "ZebraInterfaceAdd" = "EventZebraInterfaceAdd"
+
+    ``msg`` argument must be an instance of
+    :py:class: `ryu.lib.packet.zebra.ZebraMessage` and used to extract the
+    attributes for the event classes.
+    """
+
+    def __init__(self, zclient, msg):
+        super(EventZebraBase, self).__init__()
+        assert isinstance(msg, zebra.ZebraMessage)
+        self.__dict__ = msg.__dict__
+        self.zclient = zclient
+
+    def __repr__(self):
+        m = ', '.join(
+            ['%s=%r' % (k, v)
+             for k, v in self.__dict__.items() if not k.startswith('_')])
+        return "%s(%s)" % (self.__class__.__name__, m)
+
+    __str__ = __repr__
+
+
+def _event_name(body_cls):
+    return 'Event%s' % body_cls.__name__
+
+
+def message_to_event(zclient, msg):
+    """
+    Converts Zebra protocol message instance to Zebra protocol service
+    event instance.
+
+    If corresponding event class is not defined, returns None.
+
+    :param zclient: Zebra client instance.
+    :param msg: Zebra protocol message.
+    :return: Zebra protocol service event.
+    """
+    if not isinstance(msg, zebra.ZebraMessage):
+        return None
+
+    body_cls = zebra._ZebraMessageBody.lookup_command(msg.command)
+    ev_cls = getattr(MOD, _event_name(body_cls), None)
+    if ev_cls is None:
+        return None
+
+    return ev_cls(zclient, msg)
+
+
+def _define_event_class(body_cls):
+    name = _event_name(body_cls)
+
+    event_cls = type(name, (EventZebraBase,), {})
+    globals()[name] = event_cls
+
+    return event_cls
+
+
+def _generate_event_classes():
+    for zebra_cls in zebra.__dict__.values():
+        if (not inspect.isclass(zebra_cls)
+                or not issubclass(zebra_cls, zebra._ZebraMessageBody)
+                or zebra_cls.__name__.startswith('_')):
+            continue
+
+        ev = _define_event_class(zebra_cls)
+        # LOG.debug('Generated Zebra event: %s' % ev)
+        ZEBRA_EVENTS.append(ev)
+
+
+_generate_event_classes()