openstack/kuryr-kubernetes
Code Issues Proposed changes

Update namespace_security_groups driver to use OpenStackSDK.

Browse Source 
Implements: blueprint switch-to-openstacksdk
Change-Id: Id447b376ae8c10bc2658ba30f2b2009ea8c0cf08
This commit is contained in:
Roman Dobosz
2019-11-29 07:24:23 +01:00
parent 687642eef5
commit cb2f8fe874
2 changed files with 34 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
kuryr_kubernetes/controller/drivers/namespace_security_groups.py
View File

@@ -14,6 +14,7 @@
# under the License. # under the License.
from kuryr.lib._i18n import _ from kuryr.lib._i18n import _
from openstack import exceptions as os_exc
from oslo_config import cfg from oslo_config import cfg
from oslo_log import log as logging from oslo_log import log as logging
@@ -24,8 +25,6 @@ from kuryr_kubernetes.controller.drivers import base
from kuryr_kubernetes.controller.drivers import utils from kuryr_kubernetes.controller.drivers import utils
from kuryr_kubernetes import exceptions from kuryr_kubernetes import exceptions
from neutronclient.common import exceptions as n_exc
LOG = logging.getLogger(__name__) LOG = logging.getLogger(__name__)
namespace_sg_driver_opts = [ namespace_sg_driver_opts = [
@@ -93,7 +92,7 @@ class NamespacePodSecurityGroupsDriver(base.PodSecurityGroupsDriver):
return [cfg.CONF.namespace_sg.sg_allow_from_default] return [cfg.CONF.namespace_sg.sg_allow_from_default]
def create_namespace_sg(self, namespace, project_id, crd_spec): def create_namespace_sg(self, namespace, project_id, crd_spec):
neutron = clients.get_neutron_client() os_net = clients.get_network_client()
sg_name = "ns/" + namespace + "-sg" sg_name = "ns/" + namespace + "-sg"
# create the associated SG for the namespace # create the associated SG for the namespace
@@ -101,35 +100,24 @@ class NamespacePodSecurityGroupsDriver(base.PodSecurityGroupsDriver):
# default namespace is different from the rest # default namespace is different from the rest
# Default allows traffic from everywhere # Default allows traffic from everywhere
# The rest can be accessed from the default one # The rest can be accessed from the default one
sg = neutron.create_security_group( sg = os_net.create_security_group(name=sg_name,
{ project_id=project_id)
"security_group": { utils.tag_neutron_resources('security-groups', [sg.id])
"name": sg_name, os_net.create_security_group_rule(
"project_id": project_id direction="ingress",
} remote_ip_prefix=crd_spec['subnetCIDR'],
}).get('security_group') security_group_id=sg.id)
utils.tag_neutron_resources('security-groups', [sg['id']]) except os_exc.SDKException:
neutron.create_security_group_rule(
{
"security_group_rule": {
"direction": "ingress",
"remote_ip_prefix": crd_spec['subnetCIDR'],
"security_group_id": sg['id']
}
})
except n_exc.NeutronClientException:
LOG.exception("Error creating security group for the namespace " LOG.exception("Error creating security group for the namespace "
"%s", namespace) "%s", namespace)
raise raise
return {'sgId': sg['id']} return {'sgId': sg.id}
def delete_sg(self, sg_id): def delete_sg(self, sg_id):
neutron = clients.get_neutron_client() os_net = clients.get_network_client()
try: try:
neutron.delete_security_group(sg_id) os_net.delete_security_group(sg_id)
except n_exc.NotFound: except os_exc.SDKException:
LOG.debug("Security Group not found: %s", sg_id)
except n_exc.NeutronClientException:
LOG.exception("Error deleting security group %s.", sg_id) LOG.exception("Error deleting security group %s.", sg_id)
raise raise

57
kuryr_kubernetes/tests/unit/controller/drivers/test_namespace_security_groups.py
View File

@@ -19,8 +19,8 @@ from kuryr_kubernetes import constants
from kuryr_kubernetes.controller.drivers import namespace_security_groups from kuryr_kubernetes.controller.drivers import namespace_security_groups
from kuryr_kubernetes.tests import base as test_base from kuryr_kubernetes.tests import base as test_base
from kuryr_kubernetes.tests.unit import kuryr_fixtures as k_fix from kuryr_kubernetes.tests.unit import kuryr_fixtures as k_fix
import munch
from neutronclient.common import exceptions as n_exc from openstack import exceptions as os_exc
def get_pod_obj(): def get_pod_obj():
@@ -221,20 +221,18 @@ class TestNamespacePodSecurityGroupsDriver(test_base.TestCase):
namespace = 'test' namespace = 'test'
project_id = mock.sentinel.project_id project_id = mock.sentinel.project_id
sg = {'id': mock.sentinel.sg} sg = munch.Munch({'id': mock.sentinel.sg})
subnet_cidr = mock.sentinel.subnet_cidr subnet_cidr = mock.sentinel.subnet_cidr
crd_spec = { crd_spec = {'subnetCIDR': subnet_cidr}
'subnetCIDR': subnet_cidr os_net = self.useFixture(k_fix.MockNetworkClient()).client
} os_net.create_security_group.return_value = sg
neutron = self.useFixture(k_fix.MockNeutronClient()).client
neutron.create_security_group.return_value = {'security_group': sg}
create_sg_resp = cls.create_namespace_sg(m_driver, namespace, create_sg_resp = cls.create_namespace_sg(m_driver, namespace,
project_id, crd_spec) project_id, crd_spec)
self.assertEqual(create_sg_resp, {'sgId': sg['id']}) self.assertEqual(create_sg_resp, {'sgId': sg['id']})
neutron.create_security_group.assert_called_once() os_net.create_security_group.assert_called_once()
neutron.create_security_group_rule.assert_called_once() os_net.create_security_group_rule.assert_called_once()
def test_create_namespace_sg_exception(self): def test_create_namespace_sg_exception(self):
cls = namespace_security_groups.NamespacePodSecurityGroupsDriver cls = namespace_security_groups.NamespacePodSecurityGroupsDriver
@@ -246,47 +244,32 @@ class TestNamespacePodSecurityGroupsDriver(test_base.TestCase):
crd_spec = { crd_spec = {
'subnetCIDR': subnet_cidr 'subnetCIDR': subnet_cidr
} }
neutron = self.useFixture(k_fix.MockNeutronClient()).client os_net = self.useFixture(k_fix.MockNetworkClient()).client
neutron.create_security_group.side_effect = ( os_net.create_security_group.side_effect = os_exc.SDKException
n_exc.NeutronClientException)
self.assertRaises(n_exc.NeutronClientException, self.assertRaises(os_exc.SDKException, cls.create_namespace_sg,
cls.create_namespace_sg, m_driver, m_driver, namespace, project_id, crd_spec)
namespace, project_id, crd_spec)
neutron.create_security_group.assert_called_once() os_net.create_security_group.assert_called_once()
neutron.create_security_group_rule.assert_not_called() os_net.create_security_group_rule.assert_not_called()
def test_delete_sg(self): def test_delete_sg(self):
cls = namespace_security_groups.NamespacePodSecurityGroupsDriver cls = namespace_security_groups.NamespacePodSecurityGroupsDriver
m_driver = mock.MagicMock(spec=cls) m_driver = mock.MagicMock(spec=cls)
neutron = self.useFixture(k_fix.MockNeutronClient()).client os_net = self.useFixture(k_fix.MockNetworkClient()).client
sg_id = mock.sentinel.sg_id sg_id = mock.sentinel.sg_id
cls.delete_sg(m_driver, sg_id) cls.delete_sg(m_driver, sg_id)
neutron.delete_security_group.assert_called_once_with(sg_id) os_net.delete_security_group.assert_called_once_with(sg_id)
def test_delete_sg_exception(self): def test_delete_sg_exception(self):
cls = namespace_security_groups.NamespacePodSecurityGroupsDriver cls = namespace_security_groups.NamespacePodSecurityGroupsDriver
m_driver = mock.MagicMock(spec=cls) m_driver = mock.MagicMock(spec=cls)
neutron = self.useFixture(k_fix.MockNeutronClient()).client os_net = self.useFixture(k_fix.MockNetworkClient()).client
sg_id = mock.sentinel.sg_id sg_id = mock.sentinel.sg_id
neutron.delete_security_group.side_effect = ( os_net.delete_security_group.side_effect = os_exc.SDKException
n_exc.NeutronClientException)
self.assertRaises(n_exc.NeutronClientException, cls.delete_sg, self.assertRaises(os_exc.SDKException, cls.delete_sg, m_driver, sg_id)
m_driver, sg_id) os_net.delete_security_group.assert_called_once_with(sg_id)
neutron.delete_security_group.assert_called_once_with(sg_id)
def test_delete_sg_not_found(self):
cls = namespace_security_groups.NamespacePodSecurityGroupsDriver
m_driver = mock.MagicMock(spec=cls)
neutron = self.useFixture(k_fix.MockNeutronClient()).client
sg_id = mock.sentinel.sg_id
neutron.delete_security_group.side_effect = n_exc.NotFound
cls.delete_sg(m_driver, sg_id)
neutron.delete_security_group.assert_called_once_with(sg_id)