[Feature] Added archived flows

CHANGELOG.rst

@@ -21,6 +21,19 @@ Removed
 Security
 ========
 
+[5.2.0] - 2021-11.10
+********************
+
+Added
+=====
+- Added support for archiving overlapped and deleted flows, issue #33.
+- Updated settings to set the maximum number of archived flows per switch.
+
+Changed
+=======
+- Updated ``stored_flows`` box to stop using a random generated box, now it's named ``flows``. This is a breaking changing, but facilitates moving forward to also find for it, since it was assuming it only had one box in this namespace, which isn't the case anymore since we have to introduce a box for archived flows.
+
+
 [5.1.0] - 2021-11.08
 ********************
 

kytos.json

@@ -3,7 +3,7 @@
   "username": "kytos",
   "name": "flow_manager",
   "description": "Manage switches' flows through a REST API.",
-  "version": "5.1.0",
+  "version": "5.2.0",
   "napp_dependencies": ["kytos/of_core", "kytos/storehouse"],
   "license": "MIT",
   "url": "https://github.com/kytos/flow_manager.git",

main.py

@@ -4,7 +4,7 @@
 import itertools
 from collections import OrderedDict, defaultdict
 from copy import deepcopy
-from threading import Lock
+from threading import Lock, Thread
 
 from flask import jsonify, request
 from napps.kytos.flow_manager.match import match_flow
@@ -21,6 +21,8 @@
 
 from .exceptions import InvalidCommandError
 from .settings import (
+    ARCHIVED_MAX_FLOWS_PER_SWITCH,
+    ARCHIVED_ROTATION_DELETED,
     CONSISTENCY_COOKIE_IGNORED_RANGE,
     CONSISTENCY_TABLE_ID_IGNORED_RANGE,
     ENABLE_CONSISTENCY_CHECK,
@@ -80,6 +82,14 @@ def _valid_consistency_ignored(consistency_ignored_list):
     return True
 
 
+def new_archive_flow(dict_flow, reason):
+    """Build an archive flow given an stored dict flow."""
+    archive_flow = dict(dict_flow)
+    archive_flow["deleted_at"] = now().strftime("%Y-%m-%dT%H:%M:%S")
+    archive_flow["reason"] = reason
+    return archive_flow
+
+
 class Main(KytosNApp):
     """Main class to be used by Kytos controller."""
 
@@ -108,6 +118,7 @@ def setup(self):
         # {'flow_persistence': {'dpid_str': {cookie_val: [
         #                                     {'flow': {flow_dict}}]}}}
         self.stored_flows = {}
+        self.archived_flows = {}
         self.resent_flows = set()
 
     def execute(self):
@@ -116,7 +127,20 @@ def execute(self):
         The execute method is called by the run method of KytosNApp class.
         Users shouldn't call this method directly.
         """
-        self._load_flows()
+        self._load_flow_boxes()
+
+    def _load_flow_boxes(self) -> None:
+        """Load stored flow boxes."""
+        threads = []
+        for box_id, set_attr, data_key in (
+            ("flows", "stored_flows", "flow_persistence"),
+            ("flows_archived", "archived_flows", None),
+        ):
+            thread = Thread(target=self._load_flows, args=(box_id, set_attr, data_key))
+            threads.append(thread)
+            thread.start()
+        for thread in threads:
+            thread.join()
 
     def shutdown(self):
         """Shutdown routine of the NApp."""
@@ -228,6 +252,7 @@ def check_storehouse_consistency(self, switch):
                 f"{self.stored_flows.get(dpid, {}).get(cookie, [])}"
             )
 
+            archived_flows = []
             for installed_flow in flows:
                 if self.is_ignored(installed_flow.table_id, self.tab_id_ignored_range):
                     continue
@@ -237,6 +262,13 @@ def check_storehouse_consistency(self, switch):
                         f"Consistency check: alien flow on switch {dpid}, dpid"
                         " not indexed"
                     )
+                    archived_flows.append(
+                        new_archive_flow(
+                            {"flow": installed_flow.as_dict(include_id=False)},
+                            reason="alien",
+                        )
+                    )
+
                     flow = {"flows": [installed_flow.as_dict()]}
                     command = "delete_strict"
                     self._install_flows(command, flow, [switch], save=False)
@@ -247,25 +279,40 @@ def check_storehouse_consistency(self, switch):
 
                 if installed_flow not in stored_flows_list:
                     log.info(f"Consistency check: alien flow on switch {dpid}")
+                    archived_flows.append(
+                        new_archive_flow(
+                            {"flow": installed_flow.as_dict(include_id=False)},
+                            reason="alien",
+                        )
+                    )
+
                     flow = {"flows": [installed_flow.as_dict()]}
                     command = "delete_strict"
                     self._install_flows(command, flow, [switch], save=False)
                     log.info(
                         f"Flow forwarded to switch {dpid} to be deleted. Flow: {flow}"
                     )
 
+            if archived_flows:
+                with self._storehouse_lock:
+                    self._add_archived_flows_store(switch.id, archived_flows)
+
     # pylint: disable=attribute-defined-outside-init
-    def _load_flows(self):
-        """Load stored flows."""
+    def _load_flows(
+        self, box_id="flows", set_attr="stored_flows", data_key="flow_persistence"
+    ):
+        """Load stored flows of a box in an attribute."""
         try:
-            data = self.storehouse.get_data()["flow_persistence"]
+            data = self.storehouse.get_data(attr=self.storehouse.box_id_to_attr[box_id])
+            if data_key:
+                data = data[data_key]
             if "id" in data:
                 del data["id"]
-            self.stored_flows = data
-        except (KeyError, FileNotFoundError) as error:
-            log.debug(f"There are no flows to load: {error}")
+            setattr(self, set_attr, data)
+        except (KeyError, FileNotFoundError):
+            log.info(f"There are no flows to load from {box_id}.")
         else:
-            log.info("Flows loaded.")
+            log.info(f"Flows loaded from {box_id}.")
 
     def _del_matched_flows_store(self, flow_dict, switch):
         """Try to delete matching stored flows given a flow dict."""
@@ -280,7 +327,7 @@ def _del_matched_flows_store(self, flow_dict, switch):
             else [int(flow_dict.get("cookie", 0))]
         )
 
-        has_deleted_any_flow = False
+        archived_flows = []
         for cookie in cookies:
             stored_flows = stored_flows_box[switch.id].get(cookie, [])
             if not stored_flows:
@@ -292,6 +339,9 @@ def _del_matched_flows_store(self, flow_dict, switch):
                 # No strict match
                 if match_flow(flow_dict, version, stored_flow["flow"]):
                     deleted_flows_idxs.add(i)
+                    archived_flows.append(
+                        new_archive_flow(stored_flow, reason="delete")
+                    )
 
             if not deleted_flows_idxs:
                 continue
@@ -301,18 +351,42 @@ def _del_matched_flows_store(self, flow_dict, switch):
                 for i, flow in enumerate(stored_flows)
                 if i not in deleted_flows_idxs
             ]
-            has_deleted_any_flow = True
 
             if stored_flows:
                 stored_flows_box[switch.id][cookie] = stored_flows
             else:
                 stored_flows_box[switch.id].pop(cookie, None)
 
-        if has_deleted_any_flow:
+        if archived_flows:
             stored_flows_box["id"] = "flow_persistence"
             self.storehouse.save_flow(stored_flows_box)
             del stored_flows_box["id"]
             self.stored_flows = deepcopy(stored_flows_box)
+            self._add_archived_flows_store(switch.id, archived_flows)
+
+    def _add_archived_flows_store(
+        self,
+        dpid,
+        archived_flows,
+        max_len=ARCHIVED_MAX_FLOWS_PER_SWITCH,
+        offset_delete=ARCHIVED_ROTATION_DELETED,
+    ):
+        """Store archived flows."""
+        if not archived_flows:
+            return
+
+        if (
+            dpid in self.archived_flows
+            and len(self.archived_flows[dpid]) + len(archived_flows) > max_len
+        ):
+            self.archived_flows[dpid] = self.archived_flows[dpid][offset_delete:]
+
+        for archived_flow in archived_flows:
+            if dpid not in self.archived_flows:
+                self.archived_flows[dpid] = [archived_flow]
+            else:
+                self.archived_flows[dpid].append(archived_flow)
+        self.storehouse.save_archived_flow(self.archived_flows)
 
     def _add_flow_store(self, flow_dict, switch):
         """Try to add a flow dict in the store idempotently."""
@@ -325,6 +399,7 @@ def _add_flow_store(self, flow_dict, switch):
         if switch.id not in stored_flows_box:
             stored_flows_box[switch.id] = OrderedDict()
 
+        archived_flows = []
         if not stored_flows_box[switch.id].get(cookie):
             stored_flows_box[switch.id][cookie] = [installed_flow]
         else:
@@ -338,6 +413,9 @@ def _add_flow_store(self, flow_dict, switch):
                         match_flow(flow_dict, version, stored_flows[i]["flow"]),
                     )
                 ):
+                    archived_flows.append(
+                        new_archive_flow(stored_flows[i], reason="overlap")
+                    )
                     stored_flows_box[switch.id][cookie][i] = installed_flow
                     break
             else:
@@ -348,6 +426,8 @@ def _add_flow_store(self, flow_dict, switch):
         del stored_flows_box["id"]
         self.stored_flows = deepcopy(stored_flows_box)
 
+        self._add_archived_flows_store(switch.id, archived_flows)
+
     def _store_changed_flows(self, command, flow_dict, switch):
         """Store changed flows.
 

settings.py

@@ -11,3 +11,8 @@
 # To filter by a cookie or `table_id` range [(value1, value2)]
 CONSISTENCY_COOKIE_IGNORED_RANGE = []
 CONSISTENCY_TABLE_ID_IGNORED_RANGE = []
+
+# Maximum number of flows to archive per switch
+ARCHIVED_MAX_FLOWS_PER_SWITCH = 5000
+# Number of old flows to delete whenever 'ARCHIVED_MAX_FLOWS_PER_SWITCH' overflows
+ARCHIVED_ROTATION_DELETED = 1000

setup.py

@@ -24,7 +24,7 @@
 BASE_ENV = Path(os.environ.get("VIRTUAL_ENV", "/"))
 
 NAPP_NAME = "flow_manager"
-NAPP_VERSION = "5.1.0"
+NAPP_VERSION = "5.2.0"
 
 # Kytos var folder
 VAR_PATH = BASE_ENV / "var" / "lib" / "kytos"

storehouse.py

@@ -33,39 +33,44 @@ def __init__(self, controller):
 
         if "box" not in self.__dict__:
             self.box = None
+        if "box_archived" not in self.__dict__:
+            self.box_archived = None
+        self.box_id_to_attr = {"flows": "box", "flows_archived": "box_archived"}
+
         self.list_stored_boxes()
 
-    def get_data(self):
+    def get_data(self, attr="box"):
         """Return the persistence box data."""
         # Wait retrieve or create box in storehouse
         i = 0
-        while not self.box and i < BOX_RESTORE_ATTEMPTS:
+        while not getattr(self, attr) and i < BOX_RESTORE_ATTEMPTS:
             time.sleep(self.box_restore_timer)
             i += 1
-        if not self.box:
+        if not getattr(self, attr):
             error = "Error retrieving persistence box from storehouse."
             log.error(error)
             raise FileNotFoundError(error)
-        return self.box.data
+        return getattr(getattr(self, attr), "data")
 
-    def create_box(self):
+    def create_box(self, box_id=None):
         """Create a persistence box to store administrative changes."""
         content = {
             "namespace": self.namespace,
+            "box_id": box_id,
             "callback": self._create_box_callback,
             "data": {},
         }
         event = KytosEvent(name="kytos.storehouse.create", content=content)
         self.controller.buffers.app.put(event)
 
-    def _create_box_callback(self, _event, data, error):
+    def _create_box_callback(self, event, data, error):
         """Execute the callback to handle create_box."""
         if error:
             log.error(
                 f"Can't create persistence" f"box with namespace {self.namespace}"
             )
 
-        self.box = data
+        setattr(self, self.box_id_to_attr[event.content["box_id"]], data)
 
     def list_stored_boxes(self):
         """List all persistence box stored in storehouse."""
@@ -80,10 +85,11 @@ def list_stored_boxes(self):
 
     def _get_or_create_a_box_from_list_of_boxes(self, _event, data, _error):
         """Create a persistence box or retrieve the stored box."""
-        if data:
-            self.get_stored_box(data[0])
-        else:
-            self.create_box()
+        for box_id in self.box_id_to_attr:
+            if box_id in data:
+                self.get_stored_box(box_id)
+            else:
+                self.create_box(box_id)
 
     def get_stored_box(self, box_id):
         """Get persistence box from storehouse."""
@@ -97,12 +103,12 @@ def get_stored_box(self, box_id):
         event = KytosEvent(name=name, content=content)
         self.controller.buffers.app.put(event)
 
-    def _get_box_callback(self, _event, data, error):
+    def _get_box_callback(self, event, data, error):
         """Handle get_box method saving the box or logging with the error."""
         if error:
-            log.error("Persistence box not found.")
+            log.error(f"Persistence box {event.content['box_id']} not found.")
 
-        self.box = data
+        setattr(self, self.box_id_to_attr[event.content["box_id"]], data)
 
     def save_flow(self, flows):
         """Save flows in storehouse."""
@@ -123,3 +129,16 @@ def _save_flow_callback(self, _event, data, error):
             log.error(f"Can't update persistence box {data.box_id}.")
 
         log.info(f"Flow saved in {self.namespace}.{data.box_id}")
+
+    def save_archived_flow(self, stored_flows):
+        """Save archived flows in storehouse."""
+        self.box_archived.data = stored_flows
+        content = {
+            "namespace": self.namespace,
+            "box_id": self.box_archived.box_id,
+            "data": self.box_archived.data,
+            "callback": self._save_flow_callback,
+        }
+
+        event = KytosEvent(name="kytos.storehouse.update", content=content)
+        self.controller.buffers.app.put(event)