From cb1f04ed24d3236442aa10ceb916005c585e93f3 Mon Sep 17 00:00:00 2001 From: tingrli Date: Fri, 24 Jun 2016 16:15:34 -0700 Subject: [PATCH 1/6] Implemented new peer selection bmp destination logic. Adjusted config file as well. - Takes dest_peer_group as peer matching bmp collectors to send messages to. - Now connects multiple sockets as user can define multiple dest_peer_group. - Now listens to openbmp.parsed.peer for peer_name. --- setup.py | 5 +- src/bin/openbmp-forwarder | 85 +++++++++---- src/etc/openbmp-forwarder.yml | 52 +++++++- src/site-packages/openbmp/bmp.py | 137 +++++++++++++++++++++ src/site-packages/openbmp/bmp_consumer.py | 123 ++++++++++++++++-- src/site-packages/openbmp/forwarder_bmp.py | 65 +++++----- 6 files changed, 391 insertions(+), 76 deletions(-) create mode 100644 src/site-packages/openbmp/bmp.py diff --git a/setup.py b/setup.py index e8a5be7..5c31b79 100755 --- a/setup.py +++ b/setup.py @@ -11,5 +11,6 @@ data_files=[('etc', ['src/etc/openbmp-forwarder.yml'])], package_dir={'': 'src/site-packages'}, packages=['openbmp', 'openbmp.parsed'], - scripts=['src/bin/openbmp-forwarder'] - ) + scripts=['src/bin/openbmp-forwarder'], + install_requires=['ipaddress~=1.0.16', 'PyYAML~=3.11', 'python-snappy~=0.5', 'kafka-python~=1.2.2'] + ) diff --git a/src/bin/openbmp-forwarder b/src/bin/openbmp-forwarder index 53c5917..aeed7d2 100755 --- a/src/bin/openbmp-forwarder +++ b/src/bin/openbmp-forwarder @@ -15,6 +15,8 @@ import logging import yaml import time import signal +import re +import ipaddress from multiprocessing import Queue, Manager from openbmp.logger import LoggerThread @@ -46,6 +48,14 @@ def signal_handler(signum, frame): RUNNING = False +def log_print(content, isError=False): + if LOG: + if isError: + LOG.error(content) + else: + LOG.debug(content) + else: + print (content) def load_config(cfg_filename, LOG): """ Load and validate the configuration from YAML @@ -80,34 +90,12 @@ def load_config(cfg_filename, LOG): cfg['kafka']['group_id'] = APP_NAME cfg['kafka']['offset_reset_largest'] = False - if 'collector' in cfg: - if 'host' not in cfg['collector']: - if LOG: - LOG.error("Configuration is missing 'host' in collector section") - else: - print ("Configuration is missing 'host' in collector section") - sys.exit(2) - - if 'port' not in cfg['collector']: - if LOG: - LOG.error("Configuration is missing 'port' in collector section, using default of 5000") - else: - print ("Configuration is missing 'port' in collector section, using default of 5000") - - cfg['collector']['port'] = 5000 - - else: - if LOG: - LOG.error("Configuration is missing 'collector' section.") - else: - print ("Configuration is missing 'collector' section.") + if 'dest_peer_groups' not in cfg: + log_print("Configuration is missing 'dest_peer_groups' section.", True) sys.exit(2) if 'logging' not in cfg: - if LOG: - LOG.error("Configuration is missing 'logging' section.") - else: - print ("Configuration is missing 'logging' section.") + log_print("Configuration is missing 'logging' section.", True) sys.exit(2) except (IOError, yaml.YAMLError), e: @@ -174,6 +162,48 @@ def parse_cmd_args(argv): return cfg +def parse_peer_group(cfg): + """ + Parse dest_peer_group in config. + """ + peer_groups = cfg['dest_peer_groups'] + for i, peer_exp in enumerate(peer_groups): + name = peer_exp['name'] + log_print('Config: dest_peer_group name = {}'.format(name)) + if 'collector' not in peer_exp: + log_print('dest_peer_group {} is missing \'collector\' section.'.format(name), True) + sys.exit(2) + if type(peer_exp['regexp_hostname'] is list): + try: + peer_exp['regexp_hostname'] = [re.compile(regex) for regex in peer_exp['regexp_hostname']] + log_print('Config: compiled regexp hostname: {}'.format(peer_exp['regexp_hostname'])) + except re.error as err: + log_print('Invalid regular expression pattern: {}'.format(err), True) + sys.exit(2) + else: + log_print('Invalid regexp_hostname config: should be list.', True) + if type(peer_exp['prefix_range'] is list): + try: + peer_exp['prefix_range'] = [ipaddress.ip_network(unicode(prefix_exp)) for prefix_exp in + peer_exp['prefix_range']] + log_print('Config: parse prefix_range successful: {}'.format(peer_exp['prefix_range'])) + except Exception as err: + log_print('Invalid prefix range given: {}'.format(err), True) + sys.exit(2) + else: + log_print('Invalid prefix_range config: should be list.', True) + sys.exit(2) + if type(peer_exp['asn'] is list): + for asn in peer_exp['asn']: + if type(asn) is not int: + log_print("Invalid asn, must be int", True) + sys.exit(2) + else: + log_print('Invalid asn config: should be list.', True) + sys.exit(2) + cfg['dest_peer_groups'] = peer_groups + + def main(): """ Main entry point """ global LOG, RUNNING @@ -187,7 +217,7 @@ def main(): cfg_dict['max_queue_size'] = cfg['max_queue_size'] cfg_dict['logging'] = cfg['logging'] cfg_dict['kafka'] = cfg['kafka'] - cfg_dict['collector'] = cfg['collector'] + cfg_dict['dest_peer_groups'] = cfg['dest_peer_groups'] # Setup signal handers signal.signal(signal.SIGTERM, signal_handler) @@ -199,8 +229,11 @@ def main(): thread_logger = LoggerThread(log_queue, cfg_dict['logging']) thread_logger.start() + logging.basicConfig() LOG = logging.getLogger() + parse_peer_group(cfg_dict) + # Use manager queue to ensure no duplicates forward_queue = manager.Queue(cfg_dict['max_queue_size']) diff --git a/src/etc/openbmp-forwarder.yml b/src/etc/openbmp-forwarder.yml index a4d3d22..2e519a1 100644 --- a/src/etc/openbmp-forwarder.yml +++ b/src/etc/openbmp-forwarder.yml @@ -14,12 +14,53 @@ kafka: group_id : 'openbmp-forwarder' offset_reset_largest: False +# Peer group settings - Allow sending to multiple BMP destinations based on selected peers +# Order of matching +# Matching order is performed in the following sequence. The first match found is used. # -# Collector - Where to send the BMP forwarded messages -# -collector: - host: 10.1.1.1 - port: 5000 +# regexp_hostname - Hostname/regular expression is used first +# prefix_range - Prefix range is used second +# asn - Peer asn list +dest_peer_groups: + # name defines the value that is substituted for the variable. This provides a consistent + # mapping for different IP's and hostnames + - name: "lab" + + # You can specify which collector receives message about matched peers + collector: + host: 10.1.1.1 + port: 5000 + + # You can define a list of regexp's that match for hostname to group mapping + regexp_hostname: + - .*\.lab\..* + + # You can also define a list of prefixes that match for ip to group mapping + prefix_range: + - 10.100.100.0/24 + - 10.100.104.0/24 + + # You can define the matching to look at the peer asn. + asn: + - 100 + - 65000 + - 65001 + + # Keep this, it's the default entry + - name: "default" + + collector: + host: 10.1.1.1 + port: 5000 + + regexp_hostname: + - .* + + prefix_range: + - 0.0.0.0/0 + + asn: + - 0 # # Log settings @@ -61,7 +102,6 @@ logging: handlers: [file] propagate: no - # General/main program messages root: level: INFO diff --git a/src/site-packages/openbmp/bmp.py b/src/site-packages/openbmp/bmp.py new file mode 100644 index 0000000..e1f67fa --- /dev/null +++ b/src/site-packages/openbmp/bmp.py @@ -0,0 +1,137 @@ +"""OpenBMP MRT + + Copyright (c) 2013-2016 Cisco Systems, Inc. and others. All rights reserved. + This program and the accompanying materials are made available under the + terms of the Eclipse Public License v1.0 which accompanies this distribution, + and is available at http://www.eclipse.org/legal/epl-v10.html + + .. moduleauthor:: Tim Evens +""" +import socket + +from struct import unpack + + +def bmp_parse_peerhdr(data): + """ Parse BMP peer header + + :param data: BMP raw data - should start at peer header + + :return: dictionary defined as:: + { + type: , + dist_id: , + addr: , + asn: , + bgp_id: , + isIPv4: , + isPrePolicy + is2ByteASN + ts_secs: , + ts_usecs: + } + """ + hdr = { 'type': None, + 'dist_id': 0, + 'addr': None, + 'asn': 0, + 'bgp_id': 0, + 'isIPv4': True, + 'isPrePolicy': True, + 'is2ByteASN': False, + 'ts_secs': 0, + 'ts_usecs': 0} + + (type, flags, hdr['dist_id']) = unpack('>BBQ', data[:10]) + + if type == 0: + hdr['type'] = 'GLOBAL' + else: + hdr['type'] = 'L3VPN' + + if flags & 0x80: # V flag + hdr['isIPv4'] = False + else: + hdr['isIPv4'] = True + + if flags & 0x40: # L flag + hdr['isPrePolicy'] = False + else: + hdr['isPrePolicy'] = True + + if flags & 0x20: # A flag + hdr['is2ByteASN'] = True + else: + hdr['is2ByteASN'] = False + + if hdr['isIPv4']: + hdr['addr'] = socket.inet_ntop(socket.AF_INET, data[22:26]) + else: + hdr['addr'] = socket.inet_ntop(socket.AF_INET6, data[10:26]) + + (hdr['asn'],) = unpack('>I', data[26:30]) + + hdr['bgp_id'] = socket.inet_ntop(socket.AF_INET, data[30:34]) + + (hdr['ts_secs'], hdr['ts_usecs']) = unpack('>II', data[34:42]) + + return hdr + + +def bmp_parse_bmphdr(data): + """ Parse BMP header from message string + + :param data: RAW BMP message - should start at bmp header + + :return: dictionary defined as:: + { + version: , + length: , + type: , + } + """ + hdr = { 'version': None, + 'length': 0, + 'type': None } + + if not data: + return None + + (hdr['version'],) = unpack('B', data[:1]) + + if hdr['version'] == 3: + (hdr['length'], type) = unpack('>IB', data[1:6]) + hdr['length'] -= 6 # remove the bytes of the common header + + if type == 0: + hdr['type'] = 'ROUTE_MON' + elif type == 1: + hdr['type'] = 'STATS_REPORT' + elif type == 2: + hdr['type'] = 'PEER_DOWN' + elif type == 3: + hdr['type'] = 'PEER_UP' + elif type == 4: + hdr['type'] = 'INIT' + elif type == 5: + hdr['type'] = 'TERM' + else: + hdr['type'] = "UNKNOWN=%d" % type + + else: + self.LOG.error("Unsupported BMP version type of %d, cannot proceed" % hdr['version']) + return None + + return hdr + + +def resolveIp(addr): + """ Resolves an IP address to FQDN. + + :param addr: IPv4/v6 address to resovle + :return: FQDN or IP address if FQDN unknown/not found + """ + try: + return socket.gethostbyaddr(addr)[0] + except: + return addr diff --git a/src/site-packages/openbmp/bmp_consumer.py b/src/site-packages/openbmp/bmp_consumer.py index 78c8453..adf270b 100644 --- a/src/site-packages/openbmp/bmp_consumer.py +++ b/src/site-packages/openbmp/bmp_consumer.py @@ -12,7 +12,6 @@ import socket import kafka import kafka.common -import copy import time import re @@ -21,7 +20,8 @@ from openbmp.parsed.headers import headers as parsed_headers from openbmp.parsed.collector import collector from openbmp.parsed.router import router - +from openbmp.parsed.peer import peer +from openbmp.bmp import bmp_parse_bmphdr, bmp_parse_peerhdr class BMPMessageObject(object): """ OpenBMP consumer message object which is used between internal queues """ @@ -37,6 +37,12 @@ class BMPMessageObject(object): #: Router Name ROUTER_NAME = "" + DEST_CONFIG_NAME = "" + + DEST_COLLECTOR_HOST = "" + + DEST_COLLECTOR_PORT = 0 + class BMPConsumer(multiprocessing.Process): """ OpenBMP consumer for forwarding @@ -51,6 +57,9 @@ class BMPConsumer(multiprocessing.Process): # Memory cache of routers ROUTERS = {} + # Memory cache of peers + PEERS = {} + def __init__(self, cfg, forward_queue, log_queue): """ Constructor @@ -71,7 +80,7 @@ def run(self): self.LOG = init_mp_logger("bmp_consumer", self._log_queue) # Enable to topics/feeds - topics = ['openbmp.parsed.collector', 'openbmp.parsed.router', 'openbmp.bmp_raw'] + topics = ['openbmp.parsed.collector', 'openbmp.parsed.router', 'openbmp.parsed.peer', 'openbmp.bmp_raw'] self.LOG.info("Running bmp_consumer") @@ -136,6 +145,9 @@ def process_msg(self, msg): if msg.topic == 'openbmp.parsed.router': self.process_router_msg(hdr.getCollectorHashId(), data) + if msg.topic == 'openbmp.parsed.peer': + self.process_peer_msg(hdr.getCollectorHashId(), data) + elif msg.topic == 'openbmp.bmp_raw': self.process_bmp_raw_msg(hdr.getCollectorHashId(), hdr.getRouterHashId(), hdr.getRouterIp(), data) @@ -196,6 +208,46 @@ def process_router_msg(self, c_hash, data): self.LOG.debug("router parse error"); pass + def process_peer_msg(self, c_hash, data): + """ Process Peer message + + :param c_hash: Collector Hash ID + :param data: Message data to be consumed (should not contain headers) + """ + obj = peer() + + # Log messages + for row in data.split('\n'): + if len(row): + try: + obj.parse(row) + + peer_key = obj.getRouterHashId() + '_' + obj.getRemoteBgpId() + '_' + str(obj.getPeerRd()) + + self.LOG.info("peer: [%s] %s %s", obj.getAction(), + obj.getName(), obj.getIpAddress()) + + if obj.getAction() in ('up'): + + asn_len = 2 + if ('4 Octet ASN' in obj.getAdvCapabilities() and + '4 Octet ASN' in obj.getRecvCapabilities()): + asn_len = 4 + + # Update the peer hash/cache + # replace raw data with this one contains peer_name information + if obj.getAction() is not 'down': + self.PEERS[peer_key] = {'peer_name': obj.getName(), + 'remote_ip': obj.getRemoteIp(), + 'remote_asn': obj.getRemoteAsn(), + 'local_ip': obj.getLocalIp(), + 'local_asn': obj.getLocalAsn(), + 'asn_len': asn_len, + 'rd': obj.getPeerRd()} + except: + self.LOG.debug("peer parse error") + pass + def process_bmp_raw_msg(self, c_hash, r_hash, r_ip, data): """ Process BMP RAW message @@ -206,13 +258,62 @@ def process_bmp_raw_msg(self, c_hash, r_hash, r_ip, data): """ msg = BMPMessageObject() - try: - msg.COLLECTOR_ADMIN_ID = self.COLLECTORS['admin_id'] - msg.ROUTER_IP = self.ROUTERS['ip'] - msg.ROUTER_NAME = self.ROUTERS['name'] + # Parse the BMP header + bmp_hdrs = bmp_parse_bmphdr(data) - except: + # Parse the BMP headers + if bmp_hdrs['type'] in ('INIT','TERM'): + # TODO send to all and generate peer down pass - - msg.BMP_MSG = data - self._fwd_queue.put(msg) + else: + peer_hdr = bmp_parse_peerhdr(data[6:]) + + peer_key = r_hash + '_' + peer_hdr['bgp_id'] + '_' + str(peer_hdr['dist_id']) + + if c_hash in self.COLLECTORS: + msg.COLLECTOR_ADMIN_ID = self.COLLECTORS[c_hash]['admin_id'] + + if r_hash in self.ROUTERS: + msg.ROUTER_IP = self.ROUTERS[r_hash]['ip'] + msg.ROUTER_NAME = self.ROUTERS[r_hash]['name'] + + if peer_key not in self.PEERS: + self.PEERS[peer_key] = {'peer_name': '', + 'remote_ip': '', + 'remote_asn': '', + 'local_ip': peer_hdr['addr'], + 'local_asn': peer_hdr['asn'], + 'asn_len': 0, + 'rd': peer_hdr['dist_id']} + + current_peer = self.PEERS[peer_key] + + # Send to forward queue + self.LOG.debug('Doing match for PEER KEY: {}'.format(peer_key)) + + dest_peer_group = self.match_peer_group(current_peer) + if dest_peer_group is not None: + self.LOG.debug('PEER KEY: {} matched dest_peer_group {}'.format(peer_key, dest_peer_group['name'])) + msg.DEST_CONFIG_NAME = dest_peer_group['name'] + msg.DEST_COLLECTOR_HOST = dest_peer_group['collector']['host'] + msg.DEST_COLLECTOR_PORT = dest_peer_group['collector']['port'] + msg.BMP_MSG = data + self._fwd_queue.put(msg) + else: + self.LOG.error('PEER KEY: {} did not match any dest_peer_group!'.format(peer_key)) + + def match_peer_group(self, peer_dict): + peer_groups = self._cfg['dest_peer_groups'] + try: + for peer_exp in peer_groups: + for regex in peer_exp['regexp_hostname']: + if re.match(regex, peer_dict['peer_name']) is not None: + return peer_exp + for prefix in peer_exp['prefix_range']: + if peer_dict['local_ip'] in prefix.hosts(): + return peer_exp + if peer_dict['local_asn'] in peer_exp['asn']: + return peer_exp + return None + except Exception() as err: + self.LOG.error('Something went wrong: {}'.format(err)) diff --git a/src/site-packages/openbmp/forwarder_bmp.py b/src/site-packages/openbmp/forwarder_bmp.py index 989bb6f..1ca4fb8 100644 --- a/src/site-packages/openbmp/forwarder_bmp.py +++ b/src/site-packages/openbmp/forwarder_bmp.py @@ -21,6 +21,8 @@ class BMPWriter(multiprocessing.Process): Pops messages from forwarder queue and transmits them to remote bmp collector. """ + SOCKET_TIMEOUT = 5 + def __init__(self, cfg, forward_queue, log_queue): """ Constructor @@ -35,9 +37,7 @@ def __init__(self, cfg, forward_queue, log_queue): self._fwd_queue = forward_queue self._log_queue = log_queue self.LOG = None - self._isConnected = False - - self._sock = None + self._dest_socks = {} def run(self): """ Override """ @@ -57,9 +57,9 @@ def run(self): while not self.stopped(): # Do not pop any message unless connected - if self._isConnected: - msg = self._fwd_queue.get() - + msg = self._fwd_queue.get() + sock_arr = self._dest_socks[msg.DEST_CONFIG_NAME] + if sock_arr[1]: sent = False while not sent: sent = self.send(msg.BMP_MSG) @@ -67,30 +67,37 @@ def run(self): self.LOG.debug("Received bmp message: %s %s %s", msg.COLLECTOR_ADMIN_ID, msg.ROUTER_IP, msg.ROUTER_NAME) else: - self.LOG.info("Not connected, attempting to reconnect") - sleep(1) - self.connect() + self.LOG.info("Peer group \'{}\' not connected, attempting to reconnect".format(msg.DEST_CONFIG_NAME)) + sock_arr[1] = self.connect(msg.DEST_CONFIG_NAME) except KeyboardInterrupt: pass self.LOG.info("rewrite stopped") - def connect(self): - """ Connect to remote collector + def connect(self, peer_group_name=None): + """ Connect to remote controller - :return: True if connected, False otherwise/error + :param peer_group_name: the name of peer_group config + :return: if no peer_group_name given, return connection status """ - try: - self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - self._sock.connect((self._cfg['collector']['host'], self._cfg['collector']['port'])) - self._isConnected = True - self.LOG.info("Connected to remote collector: %s:%d", self._cfg['collector']['host'], - self._cfg['collector']['port']) - - except socket.error as msg: - self.LOG.error("Failed to connect to remote collector: %r", msg) - self._isConnected = False + for peer_group in self._cfg['dest_peer_groups']: + if peer_group_name is None or peer_group_name == peer_group['name']: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + is_connected = None + sock.settimeout(self.SOCKET_TIMEOUT) + try: + sock.connect((peer_group['collector']['host'], peer_group['collector']['port'])) + is_connected = True + self.LOG.info("Connected to remote collector: %s:%d", peer_group['collector']['host'], + peer_group['collector']['port']) + except socket.error as msg: + self.LOG.error("Failed to connect to remote collector: %r", msg) + is_connected = False + finally: + self._dest_socks[peer_group['name']] = [sock, is_connected] + if peer_group_name is not None: + return is_connected def send(self, msg): """ Send BMP message to socket. @@ -102,25 +109,21 @@ def send(self, msg): sent = False try: - self._sock.sendall(msg) + sock = self._dest_socks[msg.DEST_CONFIG_NAME][0] + sock.sendall(msg) sent = True except socket.error as msg: self.LOG.error("Failed to send message to collector: %r", msg) - self.disconnect() - sleep(1) - self.connect() return sent def disconnect(self): """ Disconnect from remote collector """ - if self._sock: - self._sock.close() - self._sock = None - - self._isConnected = False + for key, sock_arr in self._dest_socks: + sock_arr[0].close() + del self._dest_socks[key] def stop(self): self._stop.set() From edefdacd2dc26257a0d6c6cbb98796654954ad9e Mon Sep 17 00:00:00 2001 From: tingrli Date: Tue, 28 Jun 2016 11:11:08 -0700 Subject: [PATCH 2/6] Tries to generate PEER DOWN messages, meanwhile forward INIT & TERM to everyone --- src/bin/openbmp-forwarder | 3 + src/site-packages/openbmp/bmp.py | 10 +- src/site-packages/openbmp/bmp_consumer.py | 115 +++++++++++++++++++--- 3 files changed, 108 insertions(+), 20 deletions(-) diff --git a/src/bin/openbmp-forwarder b/src/bin/openbmp-forwarder index aeed7d2..05b3fc0 100755 --- a/src/bin/openbmp-forwarder +++ b/src/bin/openbmp-forwarder @@ -168,6 +168,9 @@ def parse_peer_group(cfg): """ peer_groups = cfg['dest_peer_groups'] for i, peer_exp in enumerate(peer_groups): + if 'name' not in peer_exp: + log_print('dest_peer_group with index {} is missing \'name\' section.'.format(i), True) + sys.exit(2) name = peer_exp['name'] log_print('Config: dest_peer_group name = {}'.format(name)) if 'collector' not in peer_exp: diff --git a/src/site-packages/openbmp/bmp.py b/src/site-packages/openbmp/bmp.py index e1f67fa..109e0c3 100644 --- a/src/site-packages/openbmp/bmp.py +++ b/src/site-packages/openbmp/bmp.py @@ -11,7 +11,6 @@ from struct import unpack - def bmp_parse_peerhdr(data): """ Parse BMP peer header @@ -32,6 +31,7 @@ def bmp_parse_peerhdr(data): } """ hdr = { 'type': None, + 'flags': None, 'dist_id': 0, 'addr': None, 'asn': 0, @@ -42,24 +42,24 @@ def bmp_parse_peerhdr(data): 'ts_secs': 0, 'ts_usecs': 0} - (type, flags, hdr['dist_id']) = unpack('>BBQ', data[:10]) + (type, hdr['flags'], hdr['dist_id']) = unpack('>BBQ', data[:10]) if type == 0: hdr['type'] = 'GLOBAL' else: hdr['type'] = 'L3VPN' - if flags & 0x80: # V flag + if hdr['flags'] & 0x80: # V flag hdr['isIPv4'] = False else: hdr['isIPv4'] = True - if flags & 0x40: # L flag + if hdr['flags'] & 0x40: # L flag hdr['isPrePolicy'] = False else: hdr['isPrePolicy'] = True - if flags & 0x20: # A flag + if hdr['flags'] & 0x20: # A flag hdr['is2ByteASN'] = True else: hdr['is2ByteASN'] = False diff --git a/src/site-packages/openbmp/bmp_consumer.py b/src/site-packages/openbmp/bmp_consumer.py index adf270b..437b8d0 100644 --- a/src/site-packages/openbmp/bmp_consumer.py +++ b/src/site-packages/openbmp/bmp_consumer.py @@ -14,6 +14,8 @@ import kafka.common import time import re +from datetime import datetime +from struct import pack from openbmp.logger import init_mp_logger @@ -200,7 +202,8 @@ def process_router_msg(self, c_hash, data): # Update the router hash/cache if obj.getHashId() not in self.ROUTERS: self.ROUTERS[obj.getHashId()] = {'ip': obj.getIpAddress(), - 'name': obj.getName()} + 'name': obj.getName(), + 'peer-list': []} # else: # self.ROUTERS.pop(obj.getHashId(), None) @@ -225,7 +228,7 @@ def process_peer_msg(self, c_hash, data): peer_key = obj.getRouterHashId() + '_' + obj.getRemoteBgpId() + '_' + str(obj.getPeerRd()) self.LOG.info("peer: [%s] %s %s", obj.getAction(), - obj.getName(), obj.getIpAddress()) + obj.getName(), obj.getLocalIp()) if obj.getAction() in ('up'): @@ -236,14 +239,21 @@ def process_peer_msg(self, c_hash, data): # Update the peer hash/cache # replace raw data with this one contains peer_name information - if obj.getAction() is not 'down': - self.PEERS[peer_key] = {'peer_name': obj.getName(), - 'remote_ip': obj.getRemoteIp(), - 'remote_asn': obj.getRemoteAsn(), - 'local_ip': obj.getLocalIp(), - 'local_asn': obj.getLocalAsn(), - 'asn_len': asn_len, - 'rd': obj.getPeerRd()} + if obj.getAction() != 'down': + if peer_key in self.PEERS: + self.PEERS[peer_key] = dict(( + {'peer_name': obj.getName(), + 'remote_ip': obj.getRemoteIp(), + 'remote_asn': obj.getRemoteAsn(), + 'local_ip': obj.getLocalIp(), + 'local_asn': obj.getLocalAsn(), + 'router_hash': obj.getRouterHashId(), + 'asn_len': asn_len, + 'rd': obj.getPeerRd()}.get(k, k), v) for (k, v) in + self.PEERS[peer_key].items()) + if obj.getRouterHashId() in self.ROUTERS: + if peer_key not in self.ROUTERS[obj.getRouterHashId()]['peer-list']: + self.ROUTERS[obj.getRouterHashId()]['peer-list'].append(peer_key) except: self.LOG.debug("peer parse error") pass @@ -261,10 +271,78 @@ def process_bmp_raw_msg(self, c_hash, r_hash, r_ip, data): # Parse the BMP header bmp_hdrs = bmp_parse_bmphdr(data) + # if bmp_hdrs['type'] != 'ROUTE_MON': + print bmp_hdrs, self.PEERS, self.ROUTERS + # Parse the BMP headers - if bmp_hdrs['type'] in ('INIT','TERM'): - # TODO send to all and generate peer down - pass + if bmp_hdrs['type'] in ('INIT', 'TERM'): + # Send to ALL + for dest_peer_group in self._cfg['dest_peer_groups']: + if c_hash in self.COLLECTORS: + msg.COLLECTOR_ADMIN_ID = self.COLLECTORS[c_hash]['admin_id'] + + if r_hash in self.ROUTERS: + msg.ROUTER_IP = self.ROUTERS[r_hash]['ip'] + msg.ROUTER_NAME = self.ROUTERS[r_hash]['name'] + + msg.DEST_CONFIG_NAME = dest_peer_group['name'] + msg.DEST_COLLECTOR_HOST = dest_peer_group['collector']['host'] + msg.DEST_COLLECTOR_PORT = dest_peer_group['collector']['port'] + msg.BMP_MSG = data + self._fwd_queue.put(msg) + + # Generate Peer Down notifications to send to peer groups + if bmp_hdrs['type'] == 'TERM': + + peer_list = [] + if r_hash in self.ROUTERS: + peer_list = self.ROUTERS[r_hash]['peer-list'] + else: + for key,peer in self.PEERS: + if peer['router_hash'] == r_hash: + peer_list.append(key) + + for peer_key in peer_list: + current_peer = self.PEERS[peer_key] + msg = BMPMessageObject() + hdr = {'version': 3, + 'length': 6, + 'type': 'PEER_DOWN', + 'peer_type': current_peer['type'], + 'peer_flags': current_peer['flags'], + 'peer_dist_id': current_peer['rd'], + 'peer_addr': current_peer['local_ip'], + 'peer_asn': current_peer['local_asn'], + 'peer_bgp_id': current_peer['bgp_id'], + 'ts_secs': int(time.time()), + 'ts_usecs': datetime.now().microsecond, + 'reason': 5 + } + msg.BMP_MSG = pack('>BIBBBQBBBBBBBBIBII',hdr) + if c_hash in self.COLLECTORS: + msg.COLLECTOR_ADMIN_ID = self.COLLECTORS[c_hash]['admin_id'] + + if r_hash in self.ROUTERS: + msg.ROUTER_IP = self.ROUTERS[r_hash]['ip'] + msg.ROUTER_NAME = self.ROUTERS[r_hash]['name'] + + # Send to forward queue + self.LOG.debug('Doing match for generated peer down notification with PEER KEY: {}'.format(peer_key)) + + dest_peer_group = self.match_peer_group(current_peer) + if dest_peer_group is not None: + self.LOG.debug( + 'PEER KEY: {} matched dest_peer_group {}'.format(peer_key, dest_peer_group['name'])) + msg.DEST_CONFIG_NAME = dest_peer_group['name'] + msg.DEST_COLLECTOR_HOST = dest_peer_group['collector']['host'] + msg.DEST_COLLECTOR_PORT = dest_peer_group['collector']['port'] + msg.BMP_MSG = data + self._fwd_queue.put(msg) + else: + self.LOG.error('PEER KEY: {} did not match any dest_peer_group!'.format(peer_key)) + del self.PEERS[peer_key] + del self.ROUTERS[r_hash] + else: peer_hdr = bmp_parse_peerhdr(data[6:]) @@ -278,14 +356,21 @@ def process_bmp_raw_msg(self, c_hash, r_hash, r_ip, data): msg.ROUTER_NAME = self.ROUTERS[r_hash]['name'] if peer_key not in self.PEERS: - self.PEERS[peer_key] = {'peer_name': '', + self.PEERS[peer_key] = {'type': peer_hdr['type'], + 'flags': peer_hdr['flags'], + 'bgp_id': peer_hdr['bgp_id'], + 'peer_name': '', 'remote_ip': '', 'remote_asn': '', 'local_ip': peer_hdr['addr'], 'local_asn': peer_hdr['asn'], + 'router_hash': r_hash, 'asn_len': 0, 'rd': peer_hdr['dist_id']} - + if r_hash in self.ROUTERS: + if peer_key not in self.ROUTERS[r_hash]['peer-list']: + self.ROUTERS[r_hash]['peer-list'].append(peer_key) + current_peer = self.PEERS[peer_key] # Send to forward queue From 7503a046ea7a92e8d5641ce4fefb8f43f8bd7568 Mon Sep 17 00:00:00 2001 From: tingrli Date: Tue, 28 Jun 2016 14:33:47 -0700 Subject: [PATCH 3/6] Managed to generated peer down message in bytes. Don't know if it actually will be correct yet. --- src/site-packages/openbmp/bmp.py | 5 +- src/site-packages/openbmp/bmp_consumer.py | 58 ++++++++++++----------- 2 files changed, 34 insertions(+), 29 deletions(-) diff --git a/src/site-packages/openbmp/bmp.py b/src/site-packages/openbmp/bmp.py index 109e0c3..ed5ebde 100644 --- a/src/site-packages/openbmp/bmp.py +++ b/src/site-packages/openbmp/bmp.py @@ -31,6 +31,7 @@ def bmp_parse_peerhdr(data): } """ hdr = { 'type': None, + 'raw_type': 0, 'flags': None, 'dist_id': 0, 'addr': None, @@ -42,9 +43,9 @@ def bmp_parse_peerhdr(data): 'ts_secs': 0, 'ts_usecs': 0} - (type, hdr['flags'], hdr['dist_id']) = unpack('>BBQ', data[:10]) + (hdr['raw_type'], hdr['flags'], hdr['dist_id']) = unpack('>BBQ', data[:10]) - if type == 0: + if hdr['raw_type'] == 0: hdr['type'] = 'GLOBAL' else: hdr['type'] = 'L3VPN' diff --git a/src/site-packages/openbmp/bmp_consumer.py b/src/site-packages/openbmp/bmp_consumer.py index 437b8d0..46d8ae0 100644 --- a/src/site-packages/openbmp/bmp_consumer.py +++ b/src/site-packages/openbmp/bmp_consumer.py @@ -268,13 +268,9 @@ def process_bmp_raw_msg(self, c_hash, r_hash, r_ip, data): """ msg = BMPMessageObject() - # Parse the BMP header + # Parse the BMP headers bmp_hdrs = bmp_parse_bmphdr(data) - # if bmp_hdrs['type'] != 'ROUTE_MON': - print bmp_hdrs, self.PEERS, self.ROUTERS - - # Parse the BMP headers if bmp_hdrs['type'] in ('INIT', 'TERM'): # Send to ALL for dest_peer_group in self._cfg['dest_peer_groups']: @@ -298,27 +294,28 @@ def process_bmp_raw_msg(self, c_hash, r_hash, r_ip, data): if r_hash in self.ROUTERS: peer_list = self.ROUTERS[r_hash]['peer-list'] else: - for key,peer in self.PEERS: + for key, peer in self.PEERS.iteritems(): if peer['router_hash'] == r_hash: peer_list.append(key) for peer_key in peer_list: current_peer = self.PEERS[peer_key] msg = BMPMessageObject() - hdr = {'version': 3, - 'length': 6, - 'type': 'PEER_DOWN', - 'peer_type': current_peer['type'], - 'peer_flags': current_peer['flags'], - 'peer_dist_id': current_peer['rd'], - 'peer_addr': current_peer['local_ip'], - 'peer_asn': current_peer['local_asn'], - 'peer_bgp_id': current_peer['bgp_id'], - 'ts_secs': int(time.time()), - 'ts_usecs': datetime.now().microsecond, - 'reason': 5 - } - msg.BMP_MSG = pack('>BIBBBQBBBBBBBBIBII',hdr) + print current_peer + msg.BMP_MSG = pack('>hhsbbqsisiih', 3, + 64, + 'PEER_DOWN', + current_peer['type'], + current_peer['flags'], + current_peer['rd'], + current_peer['local_ip'], + current_peer['local_asn'], + current_peer['bgp_id'], + int(time.time()), + datetime.now().microsecond, + 5 + ) + if c_hash in self.COLLECTORS: msg.COLLECTOR_ADMIN_ID = self.COLLECTORS[c_hash]['admin_id'] @@ -327,21 +324,23 @@ def process_bmp_raw_msg(self, c_hash, r_hash, r_ip, data): msg.ROUTER_NAME = self.ROUTERS[r_hash]['name'] # Send to forward queue - self.LOG.debug('Doing match for generated peer down notification with PEER KEY: {}'.format(peer_key)) + self.LOG.debug( + 'Doing match for generated peer down notification with PEER KEY: {}'.format(peer_key)) dest_peer_group = self.match_peer_group(current_peer) if dest_peer_group is not None: self.LOG.debug( - 'PEER KEY: {} matched dest_peer_group {}'.format(peer_key, dest_peer_group['name'])) + 'GENERATED PEER DOWN: {} matched dest_peer_group {}'.format(peer_key, + dest_peer_group['name'])) msg.DEST_CONFIG_NAME = dest_peer_group['name'] msg.DEST_COLLECTOR_HOST = dest_peer_group['collector']['host'] msg.DEST_COLLECTOR_PORT = dest_peer_group['collector']['port'] msg.BMP_MSG = data self._fwd_queue.put(msg) else: - self.LOG.error('PEER KEY: {} did not match any dest_peer_group!'.format(peer_key)) - del self.PEERS[peer_key] - del self.ROUTERS[r_hash] + self.LOG.error('GENERATED PEER DOWN: {} did not match any dest_peer_group!'.format(peer_key)) + self.PEERS.pop(peer_key, None) + self.ROUTERS.pop(r_hash, None) else: peer_hdr = bmp_parse_peerhdr(data[6:]) @@ -356,7 +355,7 @@ def process_bmp_raw_msg(self, c_hash, r_hash, r_ip, data): msg.ROUTER_NAME = self.ROUTERS[r_hash]['name'] if peer_key not in self.PEERS: - self.PEERS[peer_key] = {'type': peer_hdr['type'], + self.PEERS[peer_key] = {'type': peer_hdr['raw_type'], 'flags': peer_hdr['flags'], 'bgp_id': peer_hdr['bgp_id'], 'peer_name': '', @@ -370,7 +369,7 @@ def process_bmp_raw_msg(self, c_hash, r_hash, r_ip, data): if r_hash in self.ROUTERS: if peer_key not in self.ROUTERS[r_hash]['peer-list']: self.ROUTERS[r_hash]['peer-list'].append(peer_key) - + current_peer = self.PEERS[peer_key] # Send to forward queue @@ -388,6 +387,11 @@ def process_bmp_raw_msg(self, c_hash, r_hash, r_ip, data): self.LOG.error('PEER KEY: {} did not match any dest_peer_group!'.format(peer_key)) def match_peer_group(self, peer_dict): + """ + + :param peer_dict: the self.PEERS['key'] dictionary + :return: the dest_group_config matching this peer + """ peer_groups = self._cfg['dest_peer_groups'] try: for peer_exp in peer_groups: From 424bc42de1e564952e6033582c2b0e0dcb6908dd Mon Sep 17 00:00:00 2001 From: tingrli Date: Thu, 30 Jun 2016 15:22:20 -0700 Subject: [PATCH 4/6] Generated PEER DOWN now works. --- src/site-packages/openbmp/bmp_consumer.py | 89 +++++++++++----------- src/site-packages/openbmp/forwarder_bmp.py | 31 +++++--- 2 files changed, 64 insertions(+), 56 deletions(-) diff --git a/src/site-packages/openbmp/bmp_consumer.py b/src/site-packages/openbmp/bmp_consumer.py index 46d8ae0..457ea59 100644 --- a/src/site-packages/openbmp/bmp_consumer.py +++ b/src/site-packages/openbmp/bmp_consumer.py @@ -119,6 +119,7 @@ def run(self): except kafka.common.KafkaUnavailableError as err: self.LOG.error("Kafka Error: %s" % str(err)) + self._fwd_queue.put('DISCONNECT ALL') except KeyboardInterrupt: pass @@ -202,10 +203,7 @@ def process_router_msg(self, c_hash, data): # Update the router hash/cache if obj.getHashId() not in self.ROUTERS: self.ROUTERS[obj.getHashId()] = {'ip': obj.getIpAddress(), - 'name': obj.getName(), - 'peer-list': []} - # else: - # self.ROUTERS.pop(obj.getHashId(), None) + 'name': obj.getName()} except: self.LOG.debug("router parse error"); @@ -228,32 +226,30 @@ def process_peer_msg(self, c_hash, data): peer_key = obj.getRouterHashId() + '_' + obj.getRemoteBgpId() + '_' + str(obj.getPeerRd()) self.LOG.info("peer: [%s] %s %s", obj.getAction(), - obj.getName(), obj.getLocalIp()) + obj.getName(), obj.getLocalIp()) if obj.getAction() in ('up'): asn_len = 2 if ('4 Octet ASN' in obj.getAdvCapabilities() and - '4 Octet ASN' in obj.getRecvCapabilities()): + '4 Octet ASN' in obj.getRecvCapabilities()): asn_len = 4 # Update the peer hash/cache # replace raw data with this one contains peer_name information if obj.getAction() != 'down': if peer_key in self.PEERS: - self.PEERS[peer_key] = dict(( - {'peer_name': obj.getName(), - 'remote_ip': obj.getRemoteIp(), - 'remote_asn': obj.getRemoteAsn(), - 'local_ip': obj.getLocalIp(), - 'local_asn': obj.getLocalAsn(), - 'router_hash': obj.getRouterHashId(), - 'asn_len': asn_len, - 'rd': obj.getPeerRd()}.get(k, k), v) for (k, v) in + self.PEERS[peer_key] = dict(({'peer_name': obj.getName(), + 'remote_ip': obj.getRemoteIp(), + 'remote_asn': obj.getRemoteAsn(), + 'local_ip': obj.getLocalIp(), + 'local_asn': obj.getLocalAsn(), + 'router_hash': obj.getRouterHashId(), + 'asn_len': asn_len, + 'rd': obj.getPeerRd() + }.get(k, k), v) for (k, v) in self.PEERS[peer_key].items()) - if obj.getRouterHashId() in self.ROUTERS: - if peer_key not in self.ROUTERS[obj.getRouterHashId()]['peer-list']: - self.ROUTERS[obj.getRouterHashId()]['peer-list'].append(peer_key) + except: self.LOG.debug("peer parse error") pass @@ -287,34 +283,39 @@ def process_bmp_raw_msg(self, c_hash, r_hash, r_ip, data): msg.BMP_MSG = data self._fwd_queue.put(msg) - # Generate Peer Down notifications to send to peer groups if bmp_hdrs['type'] == 'TERM': - + # Generate Peer Down notifications to send to peer groups peer_list = [] - if r_hash in self.ROUTERS: - peer_list = self.ROUTERS[r_hash]['peer-list'] - else: - for key, peer in self.PEERS.iteritems(): - if peer['router_hash'] == r_hash: - peer_list.append(key) + + for key, peer in self.PEERS.iteritems(): + if peer['router_hash'] == r_hash: + peer_list.append(key) for peer_key in peer_list: current_peer = self.PEERS[peer_key] msg = BMPMessageObject() - print current_peer - msg.BMP_MSG = pack('>hhsbbqsisiih', 3, - 64, - 'PEER_DOWN', - current_peer['type'], - current_peer['flags'], - current_peer['rd'], - current_peer['local_ip'], - current_peer['local_asn'], - current_peer['bgp_id'], - int(time.time()), - datetime.now().microsecond, - 5 - ) + + peer_header = pack('>BBQ', current_peer['type'], current_peer['flags'], current_peer['rd']) + + if current_peer['isIPv4']: + peer_header += socket.inet_pton(socket.AF_INET, current_peer['local_ip']) + pack( + '>BBBBBBBBBBBB', 0, 0, + 0, 0, 0, 0, 0, 0, 0, 0, + 0, 0) + else: + peer_header += socket.inet_pton(socket.AF_INET6, current_peer['local_ip']) + + peer_header += pack('>I', current_peer['local_asn']) + + peer_header += socket.inet_pton(socket.AF_INET, current_peer['bgp_id']) + + peer_header += pack('>II', + int(time.time()), + datetime.now().microsecond) + + msg_body = pack('>B', 4) + + common_header = pack('>BIB', 3, len(peer_header) + len(msg_body) + 6, 2) if c_hash in self.COLLECTORS: msg.COLLECTOR_ADMIN_ID = self.COLLECTORS[c_hash]['admin_id'] @@ -335,11 +336,13 @@ def process_bmp_raw_msg(self, c_hash, r_hash, r_ip, data): msg.DEST_CONFIG_NAME = dest_peer_group['name'] msg.DEST_COLLECTOR_HOST = dest_peer_group['collector']['host'] msg.DEST_COLLECTOR_PORT = dest_peer_group['collector']['port'] - msg.BMP_MSG = data + msg.BMP_MSG = common_header + peer_header + msg_body self._fwd_queue.put(msg) else: self.LOG.error('GENERATED PEER DOWN: {} did not match any dest_peer_group!'.format(peer_key)) + self.PEERS.pop(peer_key, None) + self.ROUTERS.pop(r_hash, None) else: @@ -358,6 +361,7 @@ def process_bmp_raw_msg(self, c_hash, r_hash, r_ip, data): self.PEERS[peer_key] = {'type': peer_hdr['raw_type'], 'flags': peer_hdr['flags'], 'bgp_id': peer_hdr['bgp_id'], + 'isIPv4': peer_hdr['isIPv4'], 'peer_name': '', 'remote_ip': '', 'remote_asn': '', @@ -366,9 +370,6 @@ def process_bmp_raw_msg(self, c_hash, r_hash, r_ip, data): 'router_hash': r_hash, 'asn_len': 0, 'rd': peer_hdr['dist_id']} - if r_hash in self.ROUTERS: - if peer_key not in self.ROUTERS[r_hash]['peer-list']: - self.ROUTERS[r_hash]['peer-list'].append(peer_key) current_peer = self.PEERS[peer_key] diff --git a/src/site-packages/openbmp/forwarder_bmp.py b/src/site-packages/openbmp/forwarder_bmp.py index 1ca4fb8..42bf6fa 100644 --- a/src/site-packages/openbmp/forwarder_bmp.py +++ b/src/site-packages/openbmp/forwarder_bmp.py @@ -58,17 +58,25 @@ def run(self): # Do not pop any message unless connected msg = self._fwd_queue.get() - sock_arr = self._dest_socks[msg.DEST_CONFIG_NAME] - if sock_arr[1]: - sent = False - while not sent: - sent = self.send(msg.BMP_MSG) - - self.LOG.debug("Received bmp message: %s %s %s", msg.COLLECTOR_ADMIN_ID, - msg.ROUTER_IP, msg.ROUTER_NAME) + + if msg == 'DISCONNECT ALL': + self.disconnect() + else: - self.LOG.info("Peer group \'{}\' not connected, attempting to reconnect".format(msg.DEST_CONFIG_NAME)) - sock_arr[1] = self.connect(msg.DEST_CONFIG_NAME) + if msg.DEST_CONFIG_NAME not in self._dest_socks: + self.connect(msg.DEST_CONFIG_NAME) + + sock_arr = self._dest_socks[msg.DEST_CONFIG_NAME] + if sock_arr[1]: + sent = False + while not sent: + sent = self.send(msg.BMP_MSG, sock_arr[0]) + + self.LOG.debug("Received bmp message: %s %s %s", msg.COLLECTOR_ADMIN_ID, + msg.ROUTER_IP, msg.ROUTER_NAME) + else: + self.LOG.info("Peer group \'{}\' not connected, attempting to reconnect".format(msg.DEST_CONFIG_NAME)) + sock_arr[1] = self.connect(msg.DEST_CONFIG_NAME) except KeyboardInterrupt: pass @@ -99,7 +107,7 @@ def connect(self, peer_group_name=None): if peer_group_name is not None: return is_connected - def send(self, msg): + def send(self, msg, sock): """ Send BMP message to socket. :param msg: Message to send/write @@ -109,7 +117,6 @@ def send(self, msg): sent = False try: - sock = self._dest_socks[msg.DEST_CONFIG_NAME][0] sock.sendall(msg) sent = True From c406021570080c56571d7bb7aeb5e5026d1ea656 Mon Sep 17 00:00:00 2001 From: tingrli Date: Fri, 1 Jul 2016 15:20:41 -0700 Subject: [PATCH 5/6] Added listening to collector heartbeats as configured in YAML file. - generate PEER DOWNs after 1.1 * collector_heartbeat_interval - generate PEER DOWNs on router TERM --- src/bin/openbmp-forwarder | 1 + src/etc/openbmp-forwarder.yml | 6 + src/site-packages/openbmp/bmp_consumer.py | 153 ++++++++++++--------- src/site-packages/openbmp/forwarder_bmp.py | 2 +- 4 files changed, 98 insertions(+), 64 deletions(-) diff --git a/src/bin/openbmp-forwarder b/src/bin/openbmp-forwarder index 05b3fc0..a3a886a 100755 --- a/src/bin/openbmp-forwarder +++ b/src/bin/openbmp-forwarder @@ -221,6 +221,7 @@ def main(): cfg_dict['logging'] = cfg['logging'] cfg_dict['kafka'] = cfg['kafka'] cfg_dict['dest_peer_groups'] = cfg['dest_peer_groups'] + cfg_dict['collector_heartbeat_interval'] = cfg['collector_heartbeat_interval'] # Setup signal handers signal.signal(signal.SIGTERM, signal_handler) diff --git a/src/etc/openbmp-forwarder.yml b/src/etc/openbmp-forwarder.yml index 2e519a1..68542c8 100644 --- a/src/etc/openbmp-forwarder.yml +++ b/src/etc/openbmp-forwarder.yml @@ -14,6 +14,12 @@ kafka: group_id : 'openbmp-forwarder' offset_reset_largest: False +# +# The number of seconds after last collector heartbeat to determine if the collector is dead +# If a collector is dead, a series of PEER_DOWN will be generated to be sent to all corresponding dest_peer_group +# +collector_heartbeat_interval: 10 + # Peer group settings - Allow sending to multiple BMP destinations based on selected peers # Order of matching # Matching order is performed in the following sequence. The first match found is used. diff --git a/src/site-packages/openbmp/bmp_consumer.py b/src/site-packages/openbmp/bmp_consumer.py index 457ea59..14e7bfe 100644 --- a/src/site-packages/openbmp/bmp_consumer.py +++ b/src/site-packages/openbmp/bmp_consumer.py @@ -16,7 +16,7 @@ import re from datetime import datetime from struct import pack - +from threading import Timer from openbmp.logger import init_mp_logger from openbmp.parsed.headers import headers as parsed_headers @@ -175,7 +175,13 @@ def process_collector_msg(self, c_hash, data): # Update collector hash/cache if obj.getHashId() not in self.COLLECTORS: self.COLLECTORS[obj.getHashId()] = {'admin_id': obj.getAdminId()} - + if obj.getAction() == 'heartbeat': + cur_collector = self.COLLECTORS[obj.getHashId()] + expire_time = self._cfg['collector_heartbeat_interval'] * 1.1 + if 'timer' in cur_collector: + cur_collector['timer'].cancel() + cur_collector['timer'] = Timer(expire_time, self.handle_collector_down, [obj.getHashId()]) + cur_collector['timer'].start() else: self.COLLECTORS.pop(obj.getHashId(), None) @@ -203,7 +209,9 @@ def process_router_msg(self, c_hash, data): # Update the router hash/cache if obj.getHashId() not in self.ROUTERS: self.ROUTERS[obj.getHashId()] = {'ip': obj.getIpAddress(), - 'name': obj.getName()} + 'name': obj.getName(), + 'hash': obj.getHashId(), + 'collector_hash': c_hash} except: self.LOG.debug("router parse error"); @@ -284,66 +292,7 @@ def process_bmp_raw_msg(self, c_hash, r_hash, r_ip, data): self._fwd_queue.put(msg) if bmp_hdrs['type'] == 'TERM': - # Generate Peer Down notifications to send to peer groups - peer_list = [] - - for key, peer in self.PEERS.iteritems(): - if peer['router_hash'] == r_hash: - peer_list.append(key) - - for peer_key in peer_list: - current_peer = self.PEERS[peer_key] - msg = BMPMessageObject() - - peer_header = pack('>BBQ', current_peer['type'], current_peer['flags'], current_peer['rd']) - - if current_peer['isIPv4']: - peer_header += socket.inet_pton(socket.AF_INET, current_peer['local_ip']) + pack( - '>BBBBBBBBBBBB', 0, 0, - 0, 0, 0, 0, 0, 0, 0, 0, - 0, 0) - else: - peer_header += socket.inet_pton(socket.AF_INET6, current_peer['local_ip']) - - peer_header += pack('>I', current_peer['local_asn']) - - peer_header += socket.inet_pton(socket.AF_INET, current_peer['bgp_id']) - - peer_header += pack('>II', - int(time.time()), - datetime.now().microsecond) - - msg_body = pack('>B', 4) - - common_header = pack('>BIB', 3, len(peer_header) + len(msg_body) + 6, 2) - - if c_hash in self.COLLECTORS: - msg.COLLECTOR_ADMIN_ID = self.COLLECTORS[c_hash]['admin_id'] - - if r_hash in self.ROUTERS: - msg.ROUTER_IP = self.ROUTERS[r_hash]['ip'] - msg.ROUTER_NAME = self.ROUTERS[r_hash]['name'] - - # Send to forward queue - self.LOG.debug( - 'Doing match for generated peer down notification with PEER KEY: {}'.format(peer_key)) - - dest_peer_group = self.match_peer_group(current_peer) - if dest_peer_group is not None: - self.LOG.debug( - 'GENERATED PEER DOWN: {} matched dest_peer_group {}'.format(peer_key, - dest_peer_group['name'])) - msg.DEST_CONFIG_NAME = dest_peer_group['name'] - msg.DEST_COLLECTOR_HOST = dest_peer_group['collector']['host'] - msg.DEST_COLLECTOR_PORT = dest_peer_group['collector']['port'] - msg.BMP_MSG = common_header + peer_header + msg_body - self._fwd_queue.put(msg) - else: - self.LOG.error('GENERATED PEER DOWN: {} did not match any dest_peer_group!'.format(peer_key)) - - self.PEERS.pop(peer_key, None) - - self.ROUTERS.pop(r_hash, None) + self.handle_router_down(c_hash, r_hash) else: peer_hdr = bmp_parse_peerhdr(data[6:]) @@ -407,3 +356,81 @@ def match_peer_group(self, peer_dict): return None except Exception() as err: self.LOG.error('Something went wrong: {}'.format(err)) + + def handle_collector_down(self, c_hash): + """ Generate PEER DOWN messages for the peers under routers which ones are under this collector + + :param c_hash: The collector which is down + """ + for key in self.ROUTERS: + if self.ROUTERS[key]['collector_hash'] == c_hash: + self.handle_router_down(c_hash, key) + self.COLLECTORS.pop(c_hash, None) + + def handle_router_down(self, c_hash, r_hash): + """ Generate PEER DOWN messages for the peers under a certain router + + :param c_hash: the collector hash which the router TERM comes from + :param r_hash: the TERMed router + """ + # Generate Peer Down notifications to send to peer groups + peer_list = [] + + for key, peer in self.PEERS.iteritems(): + if peer['router_hash'] == r_hash: + peer_list.append(key) + + for peer_key in peer_list: + if peer_key in self.PEERS: + current_peer = self.PEERS[peer_key] + msg = BMPMessageObject() + + peer_header = pack('>BBQ', current_peer['type'], current_peer['flags'], current_peer['rd']) + + if current_peer['isIPv4']: + peer_header += socket.inet_pton(socket.AF_INET, current_peer['local_ip']) + pack( + '>BBBBBBBBBBBB', 0, 0, + 0, 0, 0, 0, 0, 0, 0, 0, + 0, 0) + else: + peer_header += socket.inet_pton(socket.AF_INET6, current_peer['local_ip']) + + peer_header += pack('>I', current_peer['local_asn']) + + peer_header += socket.inet_pton(socket.AF_INET, current_peer['bgp_id']) + + peer_header += pack('>II', + int(time.time()), + datetime.now().microsecond) + + msg_body = pack('>B', 4) + + common_header = pack('>BIB', 3, len(peer_header) + len(msg_body) + 6, 2) + + if c_hash in self.COLLECTORS: + msg.COLLECTOR_ADMIN_ID = self.COLLECTORS[c_hash]['admin_id'] + + if r_hash in self.ROUTERS: + msg.ROUTER_IP = self.ROUTERS[r_hash]['ip'] + msg.ROUTER_NAME = self.ROUTERS[r_hash]['name'] + + # Send to forward queue + self.LOG.debug( + 'Doing match for generated peer down notification with PEER KEY: {}'.format(peer_key)) + + dest_peer_group = self.match_peer_group(current_peer) + if dest_peer_group is not None: + self.LOG.debug( + 'GENERATED PEER DOWN: {} matched dest_peer_group {}'.format(peer_key, + dest_peer_group['name'])) + msg.DEST_CONFIG_NAME = dest_peer_group['name'] + msg.DEST_COLLECTOR_HOST = dest_peer_group['collector']['host'] + msg.DEST_COLLECTOR_PORT = dest_peer_group['collector']['port'] + msg.BMP_MSG = common_header + peer_header + msg_body + self._fwd_queue.put(msg) + else: + self.LOG.error('GENERATED PEER DOWN: {} did not match any dest_peer_group!'.format(peer_key)) + + self.PEERS.pop(peer_key, None) + + self.ROUTERS.pop(r_hash, None) \ No newline at end of file diff --git a/src/site-packages/openbmp/forwarder_bmp.py b/src/site-packages/openbmp/forwarder_bmp.py index 42bf6fa..3476398 100644 --- a/src/site-packages/openbmp/forwarder_bmp.py +++ b/src/site-packages/openbmp/forwarder_bmp.py @@ -87,7 +87,7 @@ def connect(self, peer_group_name=None): """ Connect to remote controller :param peer_group_name: the name of peer_group config - :return: if no peer_group_name given, return connection status + :return: if a peer_group_name is given, return connection status on that one """ for peer_group in self._cfg['dest_peer_groups']: if peer_group_name is None or peer_group_name == peer_group['name']: From 305d30cbd5831084b4575be6e550a7186231fb4c Mon Sep 17 00:00:00 2001 From: tingrli Date: Tue, 5 Jul 2016 10:08:07 -0700 Subject: [PATCH 6/6] Configured default heartbeat interval to be 5000. Added more to comments. --- src/etc/openbmp-forwarder.yml | 3 ++- src/site-packages/openbmp/forwarder_bmp.py | 3 ++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/src/etc/openbmp-forwarder.yml b/src/etc/openbmp-forwarder.yml index 68542c8..24a2f70 100644 --- a/src/etc/openbmp-forwarder.yml +++ b/src/etc/openbmp-forwarder.yml @@ -16,9 +16,10 @@ kafka: # # The number of seconds after last collector heartbeat to determine if the collector is dead +# Collector is considered dead after 1.1*THIS VALUE(int) after last heartbeat # If a collector is dead, a series of PEER_DOWN will be generated to be sent to all corresponding dest_peer_group # -collector_heartbeat_interval: 10 +collector_heartbeat_interval: 5000 # Peer group settings - Allow sending to multiple BMP destinations based on selected peers # Order of matching diff --git a/src/site-packages/openbmp/forwarder_bmp.py b/src/site-packages/openbmp/forwarder_bmp.py index 3476398..c5d3386 100644 --- a/src/site-packages/openbmp/forwarder_bmp.py +++ b/src/site-packages/openbmp/forwarder_bmp.py @@ -86,7 +86,8 @@ def run(self): def connect(self, peer_group_name=None): """ Connect to remote controller - :param peer_group_name: the name of peer_group config + :param peer_group_name: the name of peer_group config. If this argument is not passed, connect all socks found + in config. :return: if a peer_group_name is given, return connection status on that one """ for peer_group in self._cfg['dest_peer_groups']: