ovs plugin: automatically collect stats for external IPs

neutron/agent/linux/ovs_lib.py

@@ -444,6 +444,19 @@ def get_local_port_mac(self):
             msg = _('Unable to determine mac address for %s') % self.br_name
             raise Exception(msg)
 
+    def add_netflow(self, server, port, active_timeout, engine_id):
+        if self.bridge_exists(self.br_name):
+            self.run_vsctl(
+                ["set", "Bridge", self.br_name, "netflow=@nf", "--",
+                 "--id=@nf", "create", "NetFlow",
+                 "targets=%s\\:%s" % (server, port),
+                 "active_timeout=%s" % active_timeout,
+                 "engine_id=%s" % engine_id])
+
+    def clear_netflow(self):
+        if self.bridge_exists(self.br_name):
+            self.run_vsctl(["clear", "Bridge", self.br_name, "netflow"])
+
     def __enter__(self):
         self.create()
         return self

neutron/plugins/openvswitch/agent/netflow_agent.py

@@ -0,0 +1,255 @@
+# Copyright 2017 Eayun, 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.
+
+import socket
+import struct
+import threading
+import time
+import SocketServer
+
+import eventlet
+eventlet.monkey_patch()
+
+from neutron.agent.linux import ovs_lib
+from neutron.common import rpc as n_rpc
+from neutron.common import utils
+from neutron.openstack.common import log as logging
+from neutron.openstack.common import loopingcall
+
+LOG = logging.getLogger(__name__)
+
+NETFLOW_V5_HDR_LENGTH = 24
+NETFLOW_V5_MSG_LENGTH = 48
+
+NETFLOW_DEFAULT_SERVER = '127.0.0.1'
+NETFLOW_DEFAULT_PORT = 2055
+
+LOW_LEVEL_BRIDGE_ENGINE_ID = 1
+EXT_BRIDGE_ENGINE_ID = 2
+
+STAT_DELAY_WARN_TIMEOUT = 1
+
+SEND_STAT_TOPIC = 'neutron.netflow.send'
+RECV_STAT_TOPIC = 'neutron.netflow.recv'
+
+DEFAULT_EXT_BRIDGE = 'br-ex'
+
+
+class NetflowV5Header(object):
+    def __init__(self, _version, count, _sys_uptime, unix_secs, _unix_nsecs,
+            _flow_sequence, _engine_type, engine_id, _sampling_interval):
+        self.count = count
+        self.unix_secs = unix_secs
+        self.engine_id = engine_id
+
+    @classmethod
+    def from_data(cls, data):
+        return cls(*struct.unpack('!HHIIIIBBH', data))
+
+
+class NetflowV5Message(object):
+    def __init__(self, src_ip, dst_ip, src_num, dst_num,
+                 _nexthop, input_port, _output_port,
+                 _dpkts, doctets, _first, _last, _src_port, _dst_port,
+                 _pad1, _tcp_flags, _protocol, _tos, _src_as, _dst_as,
+                 _src_mask, _dst_mask, _pad2):
+        self.src_ip = src_ip
+        self.dst_ip = dst_ip
+        self.src_num = src_num
+        self.dst_num = dst_num
+        self.input_port = input_port
+        self.doctets = doctets
+
+    @classmethod
+    def from_data(cls, data):
+        src_ip = socket.inet_ntoa(data[0:4])
+        dst_ip = socket.inet_ntoa(data[4:8])
+        return cls(src_ip, dst_ip,
+                   *struct.unpack('!IIIHHIIIIHHBBBBHHBBH', data))
+
+
+class NetflowMessageHandler(SocketServer.BaseRequestHandler):
+
+    @staticmethod
+    def get_n_messages(count, data):
+        ret = []
+        for c in range(count):
+            offset = c * NETFLOW_V5_MSG_LENGTH
+            msg_data = data[offset:offset + NETFLOW_V5_MSG_LENGTH]
+            ret.append(NetflowV5Message.from_data(msg_data))
+        return ret
+
+    def handle(self):
+        data, client_socket = self.request
+        header_data = data[:NETFLOW_V5_HDR_LENGTH]
+        data = data[NETFLOW_V5_HDR_LENGTH:]
+
+        header = NetflowV5Header.from_data(header_data)
+        messages = self.get_n_messages(header.count, data)
+
+        agent = self.server.agent
+        recv_stats = {}
+        send_stats = {}
+        for message in messages:
+            if agent.should_add_recv(message.dst_num):
+                key = message.dst_ip
+                recv_stats[key] = recv_stats.get(key, 0) + message.doctets
+            if agent.should_add_send(
+                header.engine_id, message.src_num,
+                message.src_ip, message.input_port
+            ):
+                key = message.src_ip
+                send_stats[key] = send_stats.get(key, 0) + message.doctets
+
+        if recv_stats or send_stats:
+            agent.add_netflow_stats(header.unix_secs, recv_stats, send_stats)
+
+
+class NetflowListener(SocketServer.ThreadingUDPServer, object):
+    def __init__(self, server_address, RequestHandlerClass, agent):
+        super(
+            NetflowListener, self
+        ).__init__(server_address, RequestHandlerClass)
+        self.agent = agent
+
+
+class NetflowAgent(object):
+
+    def __init__(self, context, integ_br, root_helper, bridge_mappings, conf):
+        self.context = context
+        self.root_helper = root_helper
+
+        self.low_level_bridge = conf.low_level_ingress_bridge
+        self.ext_providers = set(conf.ext_network_providers)
+        if self.ext_providers:
+            # Multiple external networks
+            self.ext_bridge = integ_br
+        else:
+            # Single external network
+            self.ext_bridge = DEFAULT_EXT_BRIDGE
+
+        self.recv_stats = {}
+        self.send_stats = {}
+        self.pool = set()
+        self.ofport_ip_mappings = {}
+
+        self.notifier = n_rpc.get_notifier('metering')
+        self.report_pool = eventlet.greenpool.GreenPool()
+        report_loop = loopingcall.FixedIntervalLoopingCall(self._report_stats)
+        report_loop.start(interval=conf.report_interval)
+
+        self._setup_netflow(conf.active_timeout)
+
+        self.netflow_listener = NetflowListener(
+            (NETFLOW_DEFAULT_SERVER, NETFLOW_DEFAULT_PORT),
+            NetflowMessageHandler, self
+        )
+        self.listen_thread = threading.Thread(
+            target=self.netflow_listener.serve_forever
+        )
+        self.listen_thread.daemon = True
+        self.listen_thread.start()
+
+    def stop(self):
+        eventlet.spawn_n(self._destroy_netflow)
+
+    def _setup_netflow(self, active_timeout):
+        low_br = ovs_lib.OVSBridge(self.low_level_bridge, self.root_helper)
+        low_br.add_netflow(NETFLOW_DEFAULT_SERVER, NETFLOW_DEFAULT_PORT,
+                           active_timeout, LOW_LEVEL_BRIDGE_ENGINE_ID)
+        ext_br = ovs_lib.OVSBridge(self.ext_bridge, self.root_helper)
+        ext_br.add_netflow(NETFLOW_DEFAULT_SERVER, NETFLOW_DEFAULT_PORT,
+                           active_timeout, EXT_BRIDGE_ENGINE_ID)
+
+    def _destroy_netflow(self):
+        low_br = ovs_lib.OVSBridge(self.low_level_bridge, self.root_helper)
+        low_br.clear_netflow()
+        ext_br = ovs_lib.OVSBridge(self.ext_bridge, self.root_helper)
+        ext_br.clear_netflow()
+
+    def port_bound(self, port, physical_network, fixed_ips):
+        if physical_network and physical_network not in self.ext_providers:
+            return
+        port_id = port.vif_id
+        of_port_num = port.ofport
+        ip_address = fixed_ips[0]['ip_address']
+        ip_num = struct.unpack('!I', socket.inet_aton(ip_address))[0]
+        self.pool.add(ip_num)
+        self.ofport_ip_mappings[of_port_num] = {
+            'id': port_id,
+            'address': ip_address,
+            'num': ip_num
+        }
+
+    def port_unbound(self, port):
+        of_port_num = port.ofport
+        info = self.ofport_ip_mappings.pop(of_port_num, None)
+        if info:
+            self.pool.discard(info['num'])
+
+    # Functions for NetFlow message handler
+
+    def _ip_num_in_pool(self, ip_num):
+        return ip_num in self.pool
+
+    def should_add_recv(self, dst_num):
+        return self._ip_num_in_pool(dst_num)
+
+    def should_add_send(self, engine_id, src_num, src_ip, input_port):
+        if engine_id != EXT_BRIDGE_ENGINE_ID:
+            return False
+        elif not self._ip_num_in_pool(src_num):
+            return False
+        elif input_port not in self.ofport_ip_mappings:
+            return False
+
+        input_port_id = self.ofport_ip_mappings[input_port]['id']
+        input_port_address = self.ofport_ip_mappings[input_port]['address']
+        if input_port_address != src_ip:
+            LOG.warn('Get data from source IP address %(data_addr) on port '
+                     '%(port_id)s with IP address %(port_addr)s.',
+                     {'data_addr': src_ip,
+                      'port_id': input_port_id,
+                      'port_addr': input_port_address})
+        return True
+
+    @utils.synchronized('netflow')
+    def add_netflow_stats(self, timestamp, recv_stats, send_stats):
+        for ip, stat in recv_stats.items():
+            self.recv_stats[ip] = self.recv_stats.get(ip, 0) + stat
+
+        for ip, stat in send_stats.items():
+            self.send_stats[ip] = self.send_stats.get(ip, 0) + stat
+
+        delay = int(time.time()) - timestamp
+        if delay > STAT_DELAY_WARN_TIMEOUT:
+            LOG.warn('Stat handling delays for %(delay)s seconds.',
+                     {'delay': delay})
+
+    @utils.synchronized('netflow')
+    def _report_stats(self):
+        for ip, octets in self.send_stats.items():
+            data = {'ip': ip, 'bytes': octets}
+            self.report_pool.spawn_n(
+                self.notifier.info, self.context, SEND_STAT_TOPIC, data)
+        for ip, octets in self.recv_stats.items():
+            data = {'ip': ip, 'bytes': octets}
+            self.report_pool.spawn_n(
+                self.notifier.info, self.context, RECV_STAT_TOPIC, data)
+
+        self.report_pool.waitall()
+
+        self.send_stats = {}
+        self.recv_stats = {}

neutron/plugins/openvswitch/agent/ovs_neutron_agent.py

@@ -25,6 +25,7 @@
 import netaddr
 from neutron.plugins.openvswitch.agent import ovs_dvr_neutron_agent
 from neutron.plugins.openvswitch.agent import openflow_ew_dvr_agent
+from neutron.plugins.openvswitch.agent import netflow_agent
 from oslo.config import cfg
 from six import moves
 
@@ -267,6 +268,14 @@ def __init__(self, integ_br, tun_br, local_ip,
         self.sg_agent = OVSSecurityGroupAgent(self.context,
                                               self.plugin_rpc,
                                               root_helper)
+        # Netflow External IP Metering
+        if cfg.CONF.NETFLOW.enabled:
+            self.nf_agent = netflow_agent.NetflowAgent(
+                self.context, integ_br, self.root_helper, self.bridge_mappings,
+                cfg.CONF.NETFLOW)
+        else:
+            self.nf_agent = None
+
         # Initialize iteration counter
         self.iter_num = 0
         self.run_daemon_loop = True
@@ -678,6 +687,9 @@ def port_bound(self, port, net_uuid,
                                         device_owner,
                                         local_vlan_id=lvm.vlan)
 
+        if self.nf_agent:
+            self.nf_agent.port_bound(port, physical_network, fixed_ips)
+
         # Do not bind a port if it's already bound
         cur_tag = self.int_br.db_get_val("Port", port.port_name, "tag")
         if cur_tag != str(lvm.vlan):
@@ -709,6 +721,8 @@ def port_unbound(self, vif_id, net_uuid=None):
             vif_port = lvm.vif_ports[vif_id]
             self.dvr_agent.unbind_port_from_dvr(vif_port,
                                                 local_vlan_id=lvm.vlan)
+            if self.nf_agent:
+                self.nf_agent.port_unbound(vif_port)
         lvm.vif_ports.pop(vif_id, None)
 
         if not lvm.vif_ports:
@@ -1526,6 +1540,8 @@ def daemon_loop(self):
     def _handle_sigterm(self, signum, frame):
         LOG.debug("Agent caught SIGTERM, quitting daemon loop.")
         self.run_daemon_loop = False
+        if self.nf_agent:
+            self.nf_agent.stop()
 
 
 def create_agent_config_map(config):

neutron/plugins/openvswitch/common/config.py

@@ -100,8 +100,23 @@
                 help=_("Make the l2 agent run in DVR mode.")),
 ]
 
+netflow_opts = [
+    cfg.BoolOpt('enabled', default=False,
+                help=_("Enable NetFlow stat metering by external IP.")),
+    cfg.StrOpt('low_level_ingress_bridge', default='',
+               help=_("The low level ingress bridge used to monitor "
+                      "incoming traffic.")),
+    cfg.ListOpt('ext_network_providers', default=[],
+                help=_("External network providers.")),
+    cfg.IntOpt('report_interval', default=60,
+               help=_("Metering report interval.")),
+    cfg.IntOpt('active_timeout', default=20,
+               help=_("Netflow record active timeout.")),
+]
+
 
 cfg.CONF.register_opts(ovs_opts, "OVS")
 cfg.CONF.register_opts(agent_opts, "AGENT")
+cfg.CONF.register_opts(netflow_opts, "NETFLOW")
 config.register_agent_state_opts_helper(cfg.CONF)
 config.register_root_helper(cfg.CONF)