Added support from learning from past BGP updates

FORTH-ICS-INSPIRE · Oct 21, 2019 · a8ec7a5 · a8ec7a5
1 parent 6f559d3
commit a8ec7a5
Show file tree

Hide file tree

Showing 4 changed files with 63 additions and 26 deletions.
diff --git a/backend/core/configuration.py b/backend/core/configuration.py
@@ -905,6 +905,16 @@ def __check_monitors(self, _monitors):
                                 raise ArtemisError(
                                     "invalid-exabgp-autoconf-flag", entry["autoconf"]
                                 )
+                elif key == "bgpstreamhist":
+                    if "dir" in info and "autoconf" in info:
+                        if info["autoconf"] == "true":
+                            info["autoconf"] = True
+                        elif info["autoconf"] == "false":
+                            del info["autoconf"]
+                        else:
+                            raise ArtemisError(
+                                "invalid-bgpostreamhist-autoconf-flag", info["autoconf"]
+                            )
 
         @staticmethod
         def __check_asns(_asns):

diff --git a/monitor/core/monitor.py b/monitor/core/monitor.py
@@ -367,17 +367,19 @@ def init_bgpstreamhist_instance(self):
                     )
                 )
                 bgpstreamhist_dir = self.monitors["bgpstreamhist"]
-                p = Popen(
-                    [
-                        "/usr/local/bin/python3",
-                        "taps/bgpstreamhist.py",
-                        "--prefixes",
-                        self.prefix_file,
-                        "--dir",
-                        bgpstreamhist_dir,
-                    ],
-                    shell=False,
-                )
+                if "dir" in self.monitors["bgpstreamhist"]:
+                    bgpstreamhist_dir = self.monitors["bgpstreamhist"]["dir"]
+                bgpstreamhist_cmd = [
+                    "/usr/local/bin/python3",
+                    "taps/bgpstreamhist.py",
+                    "--prefixes",
+                    self.prefix_file,
+                    "--dir",
+                    bgpstreamhist_dir,
+                ]
+                if "autoconf" in self.monitors["bgpstreamhist"]:
+                    bgpstreamhist_cmd.append("-a")
+                p = Popen(bgpstreamhist_cmd, shell=False)
                 self.process_ids.append(
                     (
                         "[bgpstreamhist] {} {}".format(

diff --git a/monitor/core/taps/bgpstreamhist.py b/monitor/core/taps/bgpstreamhist.py
@@ -19,16 +19,21 @@
 log = get_logger()
 
 
-def parse_bgpstreamhist_csvs(prefixes_file=None, input_dir=None):
+def parse_bgpstreamhist_csvs(prefixes_file=None, input_dir=None, autoconf=False):
 
     prefixes = load_json(prefixes_file)
     assert prefixes is not None
 
     with Connection(RABBITMQ_URI) as connection:
-        exchange = Exchange(
+        update_exchange = Exchange(
             "bgp-update", channel=connection, type="direct", durable=False
         )
-        exchange.declare()
+        update_exchange.declare()
+        autoconf_exchange = Exchange(
+            "autoconf-local", channel=connection, type="direct", durable=False
+        )
+        update_exchange.declare()
+        autoconf_exchange.declare()
         producer = Producer(connection)
         validator = mformat_validator()
         for csv_file in glob.glob("{}/*.csv".format(input_dir)):
@@ -77,9 +82,17 @@ def parse_bgpstreamhist_csvs(prefixes_file=None, input_dir=None):
                                             for msg in msgs:
                                                 key_generator(msg)
                                                 log.debug(msg)
+                                                if autoconf:
+                                                    producer.publish(
+                                                        msg,
+                                                        exchange=autoconf_exchange,
+                                                        routing_key="update",
+                                                        serializer="json",
+                                                        priority=4,
+                                                    )
                                                 producer.publish(
                                                     msg,
-                                                    exchange=exchange,
+                                                    exchange=update_exchange,
                                                     routing_key="update",
                                                     serializer="json",
                                                 )
@@ -114,12 +127,24 @@ def parse_bgpstreamhist_csvs(prefixes_file=None, input_dir=None):
         default=None,
         help="Directory with csvs to read",
     )
+    parser.add_argument(
+        "-a",
+        "--autoconf",
+        dest="autoconf",
+        action="store_true",
+        help="Use the feed from this historical route collector to build the configuration",
+    )
 
     args = parser.parse_args()
     dir_ = args.dir.rstrip("/")
+    log.info(
+        "Starting BGPstreamhist on {} for {} (auto-conf: {})".format(
+            dir_, args.prefixes_file, args.autoconf
+        )
+    )
 
     try:
-        parse_bgpstreamhist_csvs(args.prefixes_file, dir_)
+        parse_bgpstreamhist_csvs(args.prefixes_file, dir_, args.autoconf)
     except Exception:
         log.exception("exception")
     except KeyboardInterrupt:

diff --git a/monitor/core/taps/exabgp_client.py b/monitor/core/taps/exabgp_client.py
@@ -76,19 +76,20 @@ def exabgp_msg(bgp_message):
                             for msg in msgs:
                                 key_generator(msg)
                                 log.debug(msg)
+                                if self.autoconf:
+                                    producer.publish(
+                                        msg,
+                                        exchange=self.autoconf_exchange,
+                                        routing_key="update",
+                                        serializer="json",
+                                        priority=4,
+                                    )
                                 producer.publish(
                                     msg,
                                     exchange=self.update_exchange,
                                     routing_key="update",
                                     serializer="json",
                                 )
-                                producer.publish(
-                                    msg,
-                                    exchange=self.autoconf_exchange,
-                                    routing_key="update",
-                                    serializer="json",
-                                    priority=4,
-                                )
                     else:
                         log.warning("Invalid format message: {}".format(msg))
 
@@ -101,7 +102,7 @@ def exabgp_msg(bgp_message):
                 log.exception("exception")
 
     def exit(self):
-        print("Exiting ExaBGP")
+        log.info("Exiting ExaBGP")
         if self.sio is not None:
             self.sio.disconnect()
             self.sio.wait()
@@ -128,7 +129,6 @@ def exit(self):
     parser.add_argument(
         "-a",
         "--autoconf",
-        type=bool,
         dest="autoconf",
         action="store_true",
         help="Use the feed from this local route collector to build the configuration",
@@ -137,7 +137,7 @@ def exit(self):
     args = parser.parse_args()
     ping_redis(redis)
 
-    print(
+    log.info(
         "Starting ExaBGP on {} for {} (auto-conf: {})".format(
             args.host, args.prefixes_file, args.autoconf
         )