Changes based on user story US61 and US63

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']
+      )

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,51 @@ 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):
+        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:
+            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 +220,8 @@ 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']
+    cfg_dict['collector_heartbeat_interval'] = cfg['collector_heartbeat_interval']
 
     # Setup signal handers
     signal.signal(signal.SIGTERM, signal_handler)
@@ -199,8 +233,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'])
 

src/etc/openbmp-forwarder.yml

@@ -15,11 +15,59 @@ kafka:
   offset_reset_largest: False
 
 #
-# Collector - Where to send the BMP forwarded messages
+# 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:
-  host: 10.1.1.1
-  port: 5000
+collector_heartbeat_interval: 5000
+
+# 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.
+#
+#    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 +109,6 @@ logging:
       handlers: [file]
       propagate: no
 
-
     # General/main program messages
     root:
       level: INFO

src/site-packages/openbmp/bmp.py

@@ -0,0 +1,138 @@
+"""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 <tievens@cisco.com>
+"""
+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:       <string; either 'GLOBAL' or 'L3VPN'>,
+                dist_id:    <int; 64bit route distinguisher id>,
+                addr:       <string, printed form of IP>,
+                asn:        <int; asn>,
+                bgp_id:     <int; bgp ID>,
+                isIPv4:     <bool>,
+                isPrePolicy <bool; either pre or post>
+                is2ByteASN  <bool; either 2 or 4 byte ASN>
+                ts_secs:    <timestamp in unix time>,
+                ts_usecs:   <timestamp microseconds>
+            }
+    """
+    hdr = { 'type': None,
+            'raw_type': 0,
+            'flags': None,
+            'dist_id': 0,
+            'addr': None,
+            'asn': 0,
+            'bgp_id': 0,
+            'isIPv4': True,
+            'isPrePolicy': True,
+            'is2ByteASN': False,
+            'ts_secs': 0,
+            'ts_usecs': 0}
+
+    (hdr['raw_type'], hdr['flags'], hdr['dist_id']) = unpack('>BBQ', data[:10])
+
+    if hdr['raw_type'] == 0:
+        hdr['type'] = 'GLOBAL'
+    else:
+        hdr['type'] = 'L3VPN'
+
+    if hdr['flags'] & 0x80:  # V flag
+        hdr['isIPv4'] = False
+    else:
+        hdr['isIPv4'] = True
+
+    if hdr['flags'] & 0x40:  # L flag
+        hdr['isPrePolicy'] = False
+    else:
+        hdr['isPrePolicy'] = True
+
+    if hdr['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:    <int; version of bmp>,
+                length:     <int; bmp message length in bytes not including common header>,
+                type:       <string; BMP type of message>,
+            }
+    """
+    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