[Fix] Link up notifications LINK_UP_TIMER

CHANGELOG.rst

@@ -11,9 +11,11 @@ Added
 =====
 - Publish event ``kytos/topology.current`` for topology reconciliation
 - Subscribed to event ``kytos/topology.get`` to publish the current topology
+- Added ``notified_up_at`` internal reserved metadata
 
 Changed
 =======
+- Hooked ``link_status_hook_link_up_timer`` to update ``status`` accordingly.
 
 Deprecated
 ==========
@@ -23,6 +25,7 @@ Removed
 
 Fixed
 =====
+- Fixed link up to only notify when ``LINK_UP_TIMER`` has passed
 
 Security
 ========

main.py

@@ -5,6 +5,8 @@
 # pylint: disable=wrong-import-order
 
 import time
+from collections import defaultdict
+from datetime import timezone
 from threading import Lock
 from typing import List, Optional
 
@@ -14,7 +16,7 @@
 from kytos.core import KytosEvent, KytosNApp, log, rest
 from kytos.core.common import EntityStatus
 from kytos.core.exceptions import KytosLinkCreationError
-from kytos.core.helpers import listen_to
+from kytos.core.helpers import listen_to, now
 from kytos.core.interface import Interface
 from kytos.core.link import Link
 from kytos.core.switch import Switch
@@ -40,9 +42,11 @@ def setup(self):
         self.link_up_timer = getattr(settings, 'LINK_UP_TIMER',
                                      DEFAULT_LINK_UP_TIMER)
 
-        self._lock = Lock()
         self._links_lock = Lock()
+        self._links_notify_lock = defaultdict(Lock)
         self.topo_controller = self.get_topo_controller()
+        Link.register_status_func(f"{self.napp_id}_link_up_timer",
+                                  self.link_status_hook_link_up_timer)
         self.topo_controller.bootstrap_indexes()
         self.load_topology()
 
@@ -748,8 +752,41 @@ def handle_switch_maintenance_end(self, event):
                 interface.enable()
                 self.handle_link_up(interface)
 
+    def link_status_hook_link_up_timer(self, link) -> Optional[EntityStatus]:
+        """Link status hook link up timer."""
+        tnow = time.time()
+        if (
+            link.is_active()
+            and link.is_enabled()
+            and "last_status_change" in link.metadata
+            and tnow - link.metadata['last_status_change'] < self.link_up_timer
+        ):
+            return EntityStatus.DOWN
+        return None
+
+    def notify_link_up_if_status(self, link) -> None:
+        """Tries to notify link up and topology changes based on its status
+
+        Currently, it needs to wait up to a timer."""
+        time.sleep(self.link_up_timer)
+        if link.status != EntityStatus.UP:
+            return
+        with self._links_notify_lock[link.id]:
+            notified_at = link.get_metadata("notified_up_at")
+            if (
+                notified_at
+                and (now() - notified_at.replace(tzinfo=timezone.utc)).seconds
+                < self.link_up_timer
+            ):
+                return
+            key, notified_at = "notified_up_at", now()
+            link.update_metadata(key, now())
+            self.topo_controller.add_link_metadata(link.id, {key: notified_at})
+            self.notify_topology_update()
+            self.notify_link_status_change(link, reason="link up")
+
     def handle_link_up(self, interface):
-        """Notify a link is up."""
+        """Handle link up for an interface."""
         interface.activate()
         self.topo_controller.activate_interface(interface.id)
         self.notify_topology_update()
@@ -762,34 +799,14 @@ def handle_link_up(self, interface):
             other_interface = link.endpoint_a
         if other_interface.is_active() is False:
             return
-        if link.is_active() is False:
-            link.update_metadata('last_status_change', time.time())
-            link.activate()
-
-            # As each run of this method uses a different thread,
-            # there is no risk this sleep will lock the NApp.
-            time.sleep(self.link_up_timer)
-
-            last_status_change = link.get_metadata('last_status_change')
-            now = time.time()
-            if link.is_active() and \
-                    now - last_status_change >= self.link_up_timer:
-                link.update_metadata('last_status_is_active', True)
-                self.topo_controller.activate_link(link.id, last_status_change,
-                                                   last_status_is_active=True)
-                if link.status == EntityStatus.UP:
-                    self.notify_topology_update()
-                    self.notify_link_status_change(link, reason='link up')
-        else:
-            last_status_change = time.time()
-            metadata = {'last_status_change': last_status_change,
-                        'last_status_is_active': True}
-            link.extend_metadata(metadata)
-            self.topo_controller.activate_link(link.id, last_status_change,
-                                               last_status_is_active=True)
-            if link.status == EntityStatus.UP:
-                self.notify_topology_update()
-                self.notify_link_status_change(link, reason='link up')
+        metadata = {
+            'last_status_change': time.time(),
+            'last_status_is_active': True
+        }
+        link.extend_metadata(metadata)
+        link.activate()
+        self.topo_controller.activate_link(link.id, **metadata)
+        self.notify_link_up_if_status(link)
 
     @listen_to('.*.switch.interface.link_down')
     def on_interface_link_down(self, event):
@@ -880,11 +897,20 @@ def add_links(self, event):
             log.error(f'Error creating link: {err}.')
             return
 
-        if created:
-            link.update_metadata('last_status_is_active', True)
-            self.notify_link_status_change(link, reason='link up')
-            self.notify_topology_update()
-            self.topo_controller.upsert_link(link.id, link.as_dict())
+        if not created:
+            return
+
+        self.notify_topology_update()
+        if not link.is_active():
+            return
+
+        metadata = {
+            'last_status_change': time.time(),
+            'last_status_is_active': True
+        }
+        link.extend_metadata(metadata)
+        self.topo_controller.upsert_link(link.id, link.as_dict())
+        self.notify_link_up_if_status(link)
 
     @listen_to('.*.of_lldp.network_status.updated')
     def on_lldp_status_updated(self, event):

tests/unit/test_main.py

@@ -4,10 +4,12 @@
 import pytest
 import json
 import time
+from datetime import timedelta
 from unittest import TestCase
 from unittest.mock import MagicMock, create_autospec, patch
 
 from kytos.core.common import EntityStatus
+from kytos.core.helpers import now
 from kytos.core.interface import Interface
 from kytos.core.link import Link
 from kytos.core.switch import Switch
@@ -1199,29 +1201,32 @@ def test_interface_deleted(self, mock_handle_interface_link_down):
         mock_handle_interface_link_down.assert_called()
 
     @patch('napps.kytos.topology.main.Main._get_link_from_interface')
+    @patch('napps.kytos.topology.main.Main.notify_link_up_if_status')
     @patch('napps.kytos.topology.main.Main.notify_topology_update')
-    @patch('napps.kytos.topology.main.Main.notify_link_status_change')
     def test_interface_link_up(self, *args):
         """Test interface link_up."""
-        (mock_status_change, mock_topology_update,
+        (mock_notify_topology_update,
+         mock_notify_link_up_if_status,
          mock_link_from_interface) = args
 
-        now = time.time()
+        tnow = time.time()
         mock_interface_a = create_autospec(Interface)
         mock_interface_a.is_active.return_value = False
         mock_interface_b = create_autospec(Interface)
         mock_interface_b.is_active.return_value = True
         mock_link = create_autospec(Link)
-        mock_link.get_metadata.return_value = now
+        mock_link.get_metadata.return_value = tnow
         mock_link.is_active.side_effect = [False, True]
         mock_link.endpoint_a = mock_interface_a
         mock_link.endpoint_b = mock_interface_b
         mock_link_from_interface.return_value = mock_link
         mock_link.status = EntityStatus.UP
-        self.napp.link_up_timer = 1
         self.napp.handle_interface_link_up(mock_interface_a)
-        mock_topology_update.assert_called()
-        mock_status_change.assert_called()
+        mock_notify_topology_update.assert_called()
+        mock_link.extend_metadata.assert_called()
+        mock_link.activate.assert_called()
+        self.napp.topo_controller.activate_link.assert_called()
+        mock_notify_link_up_if_status.assert_called()
 
     @patch('napps.kytos.topology.main.Main._get_link_from_interface')
     @patch('napps.kytos.topology.main.Main.notify_topology_update')
@@ -1282,11 +1287,12 @@ def test_handle_link_down_not_active(self, *args):
         assert self.napp.topo_controller.deactivate_interface.call_count == 1
 
     @patch('napps.kytos.topology.main.Main._get_link_from_interface')
+    @patch('napps.kytos.topology.main.Main.notify_link_up_if_status')
     @patch('napps.kytos.topology.main.Main.notify_topology_update')
-    @patch('napps.kytos.topology.main.Main.notify_link_status_change')
     def test_handle_link_up(self, *args):
         """Test handle link up."""
-        (mock_status_change, mock_topology_update,
+        (mock_notify_topology_update,
+         mock_notify_link_up_if_status,
          mock_link_from_interface) = args
 
         mock_interface = create_autospec(Interface)
@@ -1296,8 +1302,8 @@ def test_handle_link_up(self, *args):
         self.napp.handle_link_up(mock_interface)
         assert self.napp.topo_controller.activate_link.call_count == 1
         mock_interface.activate.assert_called()
-        assert mock_topology_update.call_count == 2
-        mock_status_change.assert_called()
+        assert mock_notify_link_up_if_status.call_count == 1
+        mock_notify_topology_update.assert_called()
 
     @patch('time.sleep')
     @patch('napps.kytos.topology.main.Main._get_link_from_interface')
@@ -1319,14 +1325,12 @@ def test_handle_link_up_intf_down(self, *args):
         assert mock_topology_update.call_count == 1
         mock_status_change.assert_not_called()
 
-    @patch('napps.kytos.topology.main.Main.notify_link_status_change')
     @patch('napps.kytos.topology.main.Main._get_link_or_create')
-    @patch('napps.kytos.topology.main.Main.notify_topology_update')
+    @patch('napps.kytos.topology.main.Main.notify_link_up_if_status')
     def test_add_links(self, *args):
         """Test add_links."""
-        (mock_notify_topology_update,
-         mock_get_link_or_create,
-         mock_link_status_change) = args
+        (mock_notify_link_up_if_status,
+         mock_get_link_or_create) = args
 
         mock_link = MagicMock()
         mock_get_link_or_create.return_value = (mock_link, True)
@@ -1336,12 +1340,11 @@ def test_add_links(self, *args):
         mock_event.content = {"interface_a": mock_intf_a,
                               "interface_b": mock_intf_b}
         self.napp.add_links(mock_event)
-        mock_link.update_metadata.assert_called()
+        mock_link.extend_metadata.assert_called()
         mock_get_link_or_create.assert_called()
-        mock_notify_topology_update.assert_called()
+        mock_notify_link_up_if_status.assert_called()
         mock_intf_a.update_link.assert_called()
         mock_intf_b.update_link.assert_called()
-        mock_link_status_change.assert_called()
         mock_link.endpoint_a = mock_intf_a
         mock_link.endpoint_b = mock_intf_b
 
@@ -1536,3 +1539,45 @@ def test_handle_switch_maintenance_end(self, handle_link_up_mock):
         event.content = content
         self.napp.handle_switch_maintenance_end(event)
         self.assertEqual(handle_link_up_mock.call_count, 5)
+
+    def test_link_status_hook_link_up_timer(self) -> None:
+        """Test status hook link up timer."""
+        last_change = time.time() - self.napp.link_up_timer + 5
+        link = MagicMock(metadata={"last_status_change": last_change})
+        link.is_active.return_value = True
+        link.is_enabled.return_value = True
+        res = self.napp.link_status_hook_link_up_timer(link)
+        assert res == EntityStatus.DOWN
+
+        last_change = time.time() - self.napp.link_up_timer
+        link.metadata["last_status_change"] = last_change
+        res = self.napp.link_status_hook_link_up_timer(link)
+        assert res is None
+
+    @patch('napps.kytos.topology.main.Main.notify_link_status_change')
+    @patch('napps.kytos.topology.main.Main.notify_topology_update')
+    @patch('time.sleep')
+    def test_notify_link_up_if_status(
+        self,
+        mock_sleep,
+        mock_notify_topo,
+        mock_notify_link,
+    ) -> None:
+        """Test notify link up if status."""
+
+        link = MagicMock(status=EntityStatus.UP)
+        link.get_metadata.return_value = now()
+        assert not self.napp.notify_link_up_if_status(link)
+        link.update_metadata.assert_not_called()
+        mock_notify_topo.assert_not_called()
+        mock_notify_link.assert_not_called()
+
+        link = MagicMock(status=EntityStatus.UP)
+        link.get_metadata.return_value = now() - timedelta(seconds=60)
+        assert not self.napp.notify_link_up_if_status(link)
+        link.update_metadata.assert_called()
+        self.napp.topo_controller.add_link_metadata.assert_called()
+        mock_notify_topo.assert_called()
+        mock_notify_link.assert_called()
+
+        assert mock_sleep.call_count == 2