Merge "Remove unused wait_for_port_detach code"

Author: Zuul

octavia/network/base.py

@@ -354,20 +354,6 @@ def get_network_configs(self, load_balancer, amphora=None):
         :raises: NotFound, NetworkNotFound, SubnetNotFound, PortNotFound
         """
 
-    @abc.abstractmethod
-    def wait_for_port_detach(self, amphora):
-        """Waits for the amphora ports device_id to be unset.
-
-        This method waits for the ports on an amphora device_id
-        parameter to be '' or None which signifies that nova has
-        finished detaching the port from the instance.
-
-        :param amphora: Amphora to wait for ports to detach.
-        :returns: None
-        :raises TimeoutException: Port did not detach in interval.
-        :raises PortNotFound: Port was not found by neutron.
-        """
-
     @abc.abstractmethod
     def update_vip_sg(self, load_balancer, vip):
         """Updates the security group for a VIP

octavia/network/drivers/neutron/allowed_address_pairs.py

@@ -797,57 +797,6 @@ def get_network_configs(self, loadbalancer, amphora=None):
                                 {'amp': amp.id, 'err': str(e)})
         return amp_configs
 
-    # TODO(johnsom) This may be dead code now. Remove in failover for v2 patch
-    def wait_for_port_detach(self, amphora):
-        """Waits for the amphora ports device_id to be unset.
-
-        This method waits for the ports on an amphora device_id
-        parameter to be '' or None which signifies that nova has
-        finished detaching the port from the instance.
-
-        :param amphora: Amphora to wait for ports to detach.
-        :returns: None
-        :raises TimeoutException: Port did not detach in interval.
-        :raises PortNotFound: Port was not found by neutron.
-        """
-        interfaces = self.get_plugged_networks(compute_id=amphora.compute_id)
-
-        ports = []
-        port_detach_timeout = CONF.networking.port_detach_timeout
-        for interface_ in interfaces:
-            port = self.get_port(port_id=interface_.port_id)
-            ips = port.fixed_ips
-            lb_network = False
-            for ip in ips:
-                if ip.ip_address == amphora.lb_network_ip:
-                    lb_network = True
-            if not lb_network:
-                ports.append(port)
-
-        for port in ports:
-            try:
-                neutron_port = self.network_proxy.get_port(
-                    port.id)
-                device_id = neutron_port['device_id']
-                start = int(time.time())
-
-                while device_id:
-                    time.sleep(CONF.networking.retry_interval)
-                    neutron_port = self.network_proxy.get_port(
-                        port.id)
-                    device_id = neutron_port['device_id']
-
-                    timed_out = int(time.time()) - start >= port_detach_timeout
-
-                    if device_id and timed_out:
-                        message = ('Port %s failed to detach (device_id %s) '
-                                   'within the required time (%s s).' %
-                                   (port.id, device_id, port_detach_timeout))
-                        raise base.TimeoutException(message)
-
-            except os_exceptions.ResourceNotFound:
-                pass
-
     def delete_port(self, port_id):
         """delete a neutron port.
 

octavia/network/drivers/noop_driver/driver.py

@@ -352,10 +352,6 @@ def get_network_configs(self, loadbalancer, amphora=None):
 
         return amp_configs
 
-    def wait_for_port_detach(self, amphora):
-        LOG.debug("failover %s no-op, wait_for_port_detach, amphora id %s",
-                  self.__class__.__name__, amphora.id)
-
     def get_qos_policy(self, qos_policy_id):
         LOG.debug("Qos Policy %s no-op, get_qos_policy qos_policy_id %s",
                   self.__class__.__name__, qos_policy_id)
@@ -520,9 +516,6 @@ def plug_port(self, amphora, port):
     def get_network_configs(self, loadbalancer, amphora=None):
         return self.driver.get_network_configs(loadbalancer, amphora)
 
-    def wait_for_port_detach(self, amphora):
-        self.driver.wait_for_port_detach(amphora)
-
     def apply_qos_on_port(self, qos_id, port_id):
         self.driver.apply_qos_on_port(qos_id, port_id)
 

octavia/tests/unit/network/drivers/neutron/test_allowed_address_pairs.py

@@ -1496,74 +1496,6 @@ def test_get_network_configs(self):
         configs = self.driver.get_network_configs(load_balancer_mock)
         self.assertEqual(1, len(configs))
 
-    @mock.patch('time.sleep')
-    def test_wait_for_port_detach(self, mock_sleep):
-        amphora = data_models.Amphora(
-            id=self.AMPHORA_ID, load_balancer_id=self.LB_ID,
-            compute_id=self.COMPUTE_ID, status=self.ACTIVE,
-            lb_network_ip=self.LB_NET_IP, ha_port_id=self.HA_PORT_ID,
-            ha_ip=self.HA_IP)
-        ports = (p for p in [
-            Port(**{"fixed_ips": [{"subnet_id": self.SUBNET_ID_1,
-                                   "ip_address": self.IP_ADDRESS_1}],
-                    "id": self.FIXED_IP_ID_1,
-                    "network_id": self.NETWORK_ID_1}),
-            Port(**{"fixed_ips": [{"subnet_id": self.SUBNET_ID_2,
-                                   "ip_address": self.IP_ADDRESS_2}],
-                    "id": self.FIXED_IP_ID_2,
-                    "network_id": self.NETWORK_ID_2})])
-        show_port_1_without_device_id = Port(**{"fixed_ips": [
-            {"subnet_id": self.SUBNET_ID_1, "ip_address": self.IP_ADDRESS_1}],
-            "id": self.FIXED_IP_ID_1, "network_id": self.NETWORK_ID_1,
-            "device_id": ''})
-        show_port_2_with_device_id = Port(**{"fixed_ips": [
-            {"subnet_id": self.SUBNET_ID_2, "ip_address": self.IP_ADDRESS_2}],
-            "id": self.FIXED_IP_ID_2, "network_id": self.NETWORK_ID_2,
-            "device_id": self.DEVICE_ID})
-        show_port_2_without_device_id = Port(**{"fixed_ips": [
-            {"subnet_id": self.SUBNET_ID_2, "ip_address": self.IP_ADDRESS_2}],
-            "id": self.FIXED_IP_ID_2, "network_id": self.NETWORK_ID_2,
-            "device_id": None})
-        self.driver.network_proxy.ports.return_value = ports
-        self.driver.network_proxy.get_port = mock.MagicMock(
-            side_effect=[show_port_1_without_device_id,
-                         show_port_2_with_device_id,
-                         show_port_2_with_device_id,
-                         show_port_2_without_device_id])
-        self.driver.wait_for_port_detach(amphora)
-        self.assertEqual(1, mock_sleep.call_count)
-
-    @mock.patch('time.time')
-    @mock.patch('time.sleep')
-    def test_wait_for_port_detach_timeout(self, mock_sleep, mock_time):
-        mock_time.side_effect = [1, 2, 6]
-        conf = oslo_fixture.Config(cfg.CONF)
-        conf.config(group="networking", port_detach_timeout=5)
-        amphora = data_models.Amphora(
-            id=self.AMPHORA_ID, load_balancer_id=self.LB_ID,
-            compute_id=self.COMPUTE_ID, status=self.ACTIVE,
-            lb_network_ip=self.LB_NET_IP, ha_port_id=self.HA_PORT_ID,
-            ha_ip=self.HA_IP)
-        ports = (p for p in [
-            Port(**{"fixed_ips": [{"subnet_id": self.SUBNET_ID_1,
-                                   "ip_address": self.IP_ADDRESS_1}],
-                    "id": self.FIXED_IP_ID_1,
-                    "network_id": self.NETWORK_ID_1}),
-            Port(**{"fixed_ips": [{"subnet_id": self.SUBNET_ID_2,
-                                   "ip_address": self.IP_ADDRESS_2}],
-                    "id": self.FIXED_IP_ID_2,
-                    "network_id": self.NETWORK_ID_2})])
-        show_port_1_with_device_id = Port(**{"fixed_ips": [
-            {"subnet_id": self.SUBNET_ID_2, "ip_address": self.IP_ADDRESS_2}],
-            "id": self.FIXED_IP_ID_2, "network_id": self.NETWORK_ID_2,
-            "device_id": self.DEVICE_ID})
-        self.driver.network_proxy.ports.return_value = ports
-        self.driver.network_proxy.get_port = mock.MagicMock(
-            return_value=show_port_1_with_device_id)
-        self.assertRaises(network_base.TimeoutException,
-                          self.driver.wait_for_port_detach,
-                          amphora)
-
     def test_delete_port(self):
         PORT_ID = uuidutils.generate_uuid()