Merge pull request #424 from kytos-ng/database_load

Set timer to reduce load

CHANGELOG.rst

@@ -14,6 +14,7 @@ Added
 - EVCs now listen to ``switch.interface.(link_up|link_down|created|deleted)`` events for activation/deactivation
 - Circuits with a vlan range are supported now. The ranges follows ``list[list[int]]`` format and both UNIs vlan should have the same ranges.
 - Usage of special vlans ``"untagged"`` and ``"any"`` now send an event to each Interface.
+- Added ``UNI_STATE_CHANGE_DELAY`` which configures the time for ``mef_eline`` to wait on link state flaps and update EVCs with last updated event.
 - Added support for ``not_ownership`` to dynamic path constraints.
 
 Changed

main.py

@@ -6,6 +6,7 @@
 import pathlib
 import time
 import traceback
+from collections import defaultdict
 from threading import Lock
 from typing import Optional
 
@@ -14,7 +15,7 @@
 from kytos.core import KytosNApp, log, rest
 from kytos.core.events import KytosEvent
 from kytos.core.exceptions import KytosTagError
-from kytos.core.helpers import (alisten_to, listen_to, load_spec,
+from kytos.core.helpers import (alisten_to, listen_to, load_spec, now,
                                 validate_openapi)
 from kytos.core.interface import TAG, UNI, TAGRange
 from kytos.core.link import Link
@@ -63,6 +64,8 @@ def setup(self):
         # Every create/update/delete must be synced to mongodb.
         self.circuits = {}
 
+        self._intf_events = defaultdict(dict)
+        self._lock_interfaces = defaultdict(Lock)
         self.table_group = {"epl": 0, "evpl": 0}
         self._lock = Lock()
         self.execute_as_loop(settings.DEPLOY_EVCS_INTERVAL)
@@ -759,16 +762,42 @@ def handle_link_up(self, event):
 
     # Possibly replace this with interruptions?
     @listen_to(
-        '.*.switch.interface.(link_up|link_down|created|deleted)',
-        pool='dynamic_single'
+        '.*.switch.interface.(link_up|link_down|created|deleted)'
     )
     def on_interface_link_change(self, event: KytosEvent):
         """
-        Handler for interface link_up and link_down events
+        Handler for interface link_up and link_down events.
         """
-        with self._lock:
+        self.handle_on_interface_link_change(event)
+
+    def handle_on_interface_link_change(self, event: KytosEvent):
+        """
+        Handler to sort interface events {link_(up, down), create, deleted}
+
+        To avoid multiple database updated (link flap):
+        Every interface is identfied and processed in parallel.
+        Once an interface event is received a time is started.
+        While time is running self._intf_events will be updated.
+        After time has passed last received event will be processed.
+        """
+        iface = event.content.get("interface")
+        with self._lock_interfaces[iface.id]:
+            _now = event.timestamp
+            # Return out of order events
+            if (
+                iface.id in self._intf_events
+                and self._intf_events[iface.id]["event"].timestamp > _now
+            ):
+                return
+            self._intf_events[iface.id].update({"event": event})
+            if "last_acquired" in self._intf_events[iface.id]:
+                return
+            self._intf_events[iface.id].update({"last_acquired": now()})
+        time.sleep(settings.UNI_STATE_CHANGE_DELAY)
+        with self._lock_interfaces[iface.id]:
+            event = self._intf_events[iface.id]["event"]
+            self._intf_events[iface.id].pop('last_acquired', None)
             _, _, event_type = event.name.rpartition('.')
-            iface = event.content.get("interface")
             if event_type in ('link_up', 'created'):
                 self.handle_interface_link_up(iface)
             elif event_type in ('link_down', 'deleted'):

settings.py

@@ -55,3 +55,7 @@
 QUEUE_ID = None
 # Default spf_attribute. Allowed values: "hop", "priority", and "delay"
 SPF_ATTRIBUTE = "hop"
+
+# Time (seconds) to update EVC after interface event
+# ".*.switch.interface.(link_up|link_down|created|deleted)"
+UNI_STATE_CHANGE_DELAY = 0.1

tests/unit/test_main.py

@@ -1,9 +1,10 @@
 """Module to test the main napp file."""
-from unittest.mock import (AsyncMock, MagicMock, PropertyMock, call,
+from unittest.mock import (AsyncMock, MagicMock, Mock, PropertyMock, call,
                            create_autospec, patch)
 
 import pytest
 from kytos.lib.helpers import get_controller_mock, get_test_client
+from kytos.core.helpers import now
 from kytos.core.common import EntityStatus
 from kytos.core.events import KytosEvent
 from kytos.core.exceptions import KytosTagError
@@ -2565,3 +2566,52 @@ def test_check_no_tag_duplication(self):
 
         self.napp._check_no_tag_duplication(evc_id, None, None)
         assert evc.check_no_tag_duplicate.call_count == 3
+
+    @patch("napps.kytos.mef_eline.main.time")
+    @patch("napps.kytos.mef_eline.main.Main.handle_interface_link_up")
+    @patch("napps.kytos.mef_eline.main.Main.handle_interface_link_down")
+    def test_handle_on_interface_link_change(
+        self,
+        mock_down,
+        mock_up,
+        mock_time
+    ):
+        """Test handle_on_interface_link_change"""
+        mock_time.sleep.return_value = True
+        mock_intf = Mock()
+        mock_intf.id = "mock_intf"
+
+        # Created/link_up
+        name = '.*.switch.interface.created'
+        content = {"interface": mock_intf}
+        event = KytosEvent(name=name, content=content)
+        self.napp.handle_on_interface_link_change(event)
+        assert mock_down.call_count == 0
+        assert mock_up.call_count == 1
+
+        # Deleted/link_down
+        name = '.*.switch.interface.deleted'
+        event = KytosEvent(name=name, content=content)
+        self.napp.handle_on_interface_link_change(event)
+        assert mock_down.call_count == 1
+        assert mock_up.call_count == 1
+
+        # Event delay
+        self.napp._intf_events[mock_intf.id]["last_acquired"] = "mock_time"
+        for _ in range(1, 6):
+            self.napp.handle_on_interface_link_change(event)
+        assert mock_down.call_count == 1
+        assert mock_up.call_count == 1
+
+        self.napp._intf_events[mock_intf.id].pop("last_acquired")
+        self.napp.handle_on_interface_link_change(event)
+        assert mock_down.call_count == 2
+        assert mock_up.call_count == 1
+
+        # Out of order event
+        event = KytosEvent(name=name, content=content)
+        self.napp._intf_events[mock_intf.id]["event"] = Mock(timestamp=now())
+
+        self.napp.handle_on_interface_link_change(event)
+        assert mock_down.call_count == 2
+        assert mock_up.call_count == 1