From 6c081bfabc7a999b211cf6d70d0f2d99dcc5967a Mon Sep 17 00:00:00 2001 From: Hunt Xu Date: Wed, 27 Sep 2017 11:38:10 +0800 Subject: [PATCH] ovs plugin: automatically collect stats for external IPs Use OpenvSwitch's NetFlow support to automatically collect traffic stats to/from floating IP or router's gateway IP on external network. Recv stats are collected on both low level ingress ovs(traffic from outside) and the ovs for external network ports(traffic from ports on the same host). Send stats are only collected on the ovs for external network ports. This commit requires the use of EayunStack floating IP mechanism. Fixes: redmine #11061 Signed-off-by: Hunt Xu --- neutron/agent/linux/ovs_lib.py | 13 + .../openvswitch/agent/netflow_agent.py | 255 ++++++++++++++++++ .../openvswitch/agent/ovs_neutron_agent.py | 16 ++ neutron/plugins/openvswitch/common/config.py | 15 ++ 4 files changed, 299 insertions(+) create mode 100644 neutron/plugins/openvswitch/agent/netflow_agent.py diff --git a/neutron/agent/linux/ovs_lib.py b/neutron/agent/linux/ovs_lib.py index 2013ba170c6..67a31482805 100644 --- a/neutron/agent/linux/ovs_lib.py +++ b/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 diff --git a/neutron/plugins/openvswitch/agent/netflow_agent.py b/neutron/plugins/openvswitch/agent/netflow_agent.py new file mode 100644 index 00000000000..c80d171bf36 --- /dev/null +++ b/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 = {} diff --git a/neutron/plugins/openvswitch/agent/ovs_neutron_agent.py b/neutron/plugins/openvswitch/agent/ovs_neutron_agent.py index a0195750919..b5272e2d937 100644 --- a/neutron/plugins/openvswitch/agent/ovs_neutron_agent.py +++ b/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): diff --git a/neutron/plugins/openvswitch/common/config.py b/neutron/plugins/openvswitch/common/config.py index d6805822b12..112ee55c979 100644 --- a/neutron/plugins/openvswitch/common/config.py +++ b/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)