starlingx/openstack-armada-app
Code Issues Proposed changes

Improve NetApp utility functions

Browse Source 
This change introduces common constants across helm charts, adds new
NetApp utility functions, and improves existing ones to better support
NetApp backends configuration.

The function `check_netapp_backends` now requires two conditions for a
backend to be considered available:
1. The backend must be installed in the platform and automatically
   discovered by the app via `netapp_backends_auto_discovery()`.
2. If user overrides are defined for the given storage backend through
   the `storage_conf.storage_backends` config, the backend must be
   explicitly enabled by the user in these overrides. If it doesn't, it
   will be considered as disabled.

The new utility functions `discover_netapp_credentials` and
`discover_netapp_configs` enable the helm charts to automatically
discover the NetApp backend configs by querying them from the underlying
platform.

Test Plan:
[PASS] Build OpenStack packages and tarball
[PASS] Upload and apply tarball to a VBox AIO-SX with Ceph and NetApp
       iSCSI backends
[PASS] Verify OpenStack volume services for NetApp iSCSI and Ceph
       (`openstack volume service list`)
[PASS] Verify volume types for NetApp iSCSI and Ceph
       (`openstack volume type list`)
[PASS] Create empty NetApp iSCSI volume
[PASS] Create NetApp iSCSI volume from image
[PASS] Create backup for NetApp iSCSI volume
[PASS] Create Ceph volume from image
[PASS] Create backup for Ceph volume
[PASS] Delete volumes and backups

Story: 2011281
Task: 53477

Change-Id: I7aa3fc74236a64e850524994b8bb890061075786
Signed-off-by: Alex Figueiredo <alex.fernandesfigueiredo@windriver.com>
This commit is contained in:
Alex Figueiredo
2025-12-12 09:45:16 -03:00
parent 4e8d44e00b
commit 27ed8cf93f
7 changed files with 683 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python3-k8sapp-openstack/k8sapp_openstack/k8sapp_openstack/common/constants.py
+23 -1
View File
@@ -158,12 +158,34 @@ BACKEND_DEFAULT_STORAGE_CLASS = "general"
BACKEND_TYPE_NETAPP_NFS = "ontap-nas"
BACKEND_TYPE_NETAPP_ISCSI = "ontap-san"
BACKEND_TYPE_NETAPP_FC = "ontap-san"
NETAPP_NFS_NAS_TYPE = "nfs"
NETAPP_ISCSI_SAN_TYPE = "iscsi"
NETAPP_FC_SAN_TYPE = "fcp"
NETAPP_NFS_OPENSTACK_PROTOCOL = "nfs"
NETAPP_ISCSI_OPENSTACK_PROTOCOL = "iscsi"
NETAPP_FC_OPENSTACK_PROTOCOL = "fc"
NETAPP_SUPPORTED_BACKENDS = [
NETAPP_NFS_BACKEND_NAME,
NETAPP_ISCSI_BACKEND_NAME,
NETAPP_FC_BACKEND_NAME
]
NETAPP_BACKEND_TO_OPENSTACK_PROTOCOL = {
NETAPP_NFS_BACKEND_NAME: NETAPP_NFS_OPENSTACK_PROTOCOL,
NETAPP_ISCSI_BACKEND_NAME: NETAPP_ISCSI_OPENSTACK_PROTOCOL,
NETAPP_FC_BACKEND_NAME: NETAPP_FC_OPENSTACK_PROTOCOL,
}
NETAPP_BACKEND_TO_TYPE = {
NETAPP_NFS_BACKEND_NAME: BACKEND_TYPE_NETAPP_NFS,
NETAPP_ISCSI_BACKEND_NAME: BACKEND_TYPE_NETAPP_ISCSI,
NETAPP_FC_BACKEND_NAME: BACKEND_TYPE_NETAPP_FC,
}
# NetApp Default and Placeholder Values
NETAPP_CINDER_VOLUME_DRIVER = "cinder.volume.drivers.netapp.common.NetAppDriver"
NETAPP_STORAGE_FAMILY = "ontap_cluster"
NETAPP_DEFAULT_SERVER_HOSTNAME = "127.0.0.1"
NETAPP_DEFAULT_SERVER_PORT = 80
NETAPP_DEFAULT_SERVER_PORT = 443
NETAPP_DEFAULT_SERVER_TRANSPORT_TYPE = "https"
NETAPP_DEFAULT_LOGIN = "netapp-login"
NETAPP_DEFAULT_PASSWORD = "netapp-password"
NETAPP_DEFAULT_VSERVER = "netapp-vserver"
python3-k8sapp-openstack/k8sapp_openstack/k8sapp_openstack/helm/cinder.py
+10 -19
View File
@@ -112,9 +112,9 @@ class CinderHelm(openstack.OpenstackBaseHelm):
cinder_backup_privileged = False
if is_netapp_available():
netapp_backends = check_netapp_backends()
self._netapp_nfs_enabled = netapp_backends["nfs"]
self._netapp_iscsi_enabled = netapp_backends["iscsi"]
self._netapp_fc_enabled = netapp_backends["fc"]
self._netapp_nfs_enabled = netapp_backends[app_constants.NETAPP_NFS_BACKEND_NAME]
self._netapp_iscsi_enabled = netapp_backends[app_constants.NETAPP_ISCSI_BACKEND_NAME]
self._netapp_fc_enabled = netapp_backends[app_constants.NETAPP_FC_BACKEND_NAME]
# If NetApp is using NFS, the cinder-volume pod cannot have a readOnly filesystem,
# as the NFS will be mounted into the pod during initialization
@@ -132,7 +132,8 @@ class CinderHelm(openstack.OpenstackBaseHelm):
# Update priority map for netapp backend
# First we select only the enabled netapp backends
netapp_enabled_backends = [
f"netapp-{b}" for b in netapp_backends.keys() if netapp_backends[b]]
b for b in netapp_backends.keys() if netapp_backends[b]
]
# Then we set the value in map accordingly to the position in the priority list
for backend in netapp_enabled_backends:
self.priority_map[backend] = self.VOLUME_PRIORITY_LIST.index(backend) if (
@@ -332,14 +333,9 @@ class CinderHelm(openstack.OpenstackBaseHelm):
# Get available NetApp backends
netapp_backends = check_netapp_backends()
netapp_array = [
app_constants.NETAPP_NFS_BACKEND_NAME if netapp_backends.get("nfs") else None,
app_constants.NETAPP_ISCSI_BACKEND_NAME if netapp_backends.get("iscsi") else None,
app_constants.NETAPP_FC_BACKEND_NAME if netapp_backends.get("fc") else None,
b for b in netapp_backends.keys() if netapp_backends[b]
]
# Remove None values
netapp_array = [item for item in netapp_array if item]
# Add NetApp backends to Cinder enabled_backends list, ensuring no duplicates
existing_backends = cinder_overrides['DEFAULT'].get('enabled_backends', '').split(',')
backends_list = list(filter(None, set(existing_backends + netapp_array)))
@@ -395,14 +391,9 @@ class CinderHelm(openstack.OpenstackBaseHelm):
# Get available NetApp backends
netapp_backends = check_netapp_backends()
netapp_array = [
app_constants.NETAPP_NFS_BACKEND_NAME if netapp_backends.get("nfs") else None,
app_constants.NETAPP_ISCSI_BACKEND_NAME if netapp_backends.get("iscsi") else None,
app_constants.NETAPP_FC_BACKEND_NAME if netapp_backends.get("fc") else None,
b for b in netapp_backends.keys() if netapp_backends[b]
]
# Remove None values
netapp_array = [item for item in netapp_array if item]
common_netapp_config = {
'volume_driver': app_constants.NETAPP_CINDER_VOLUME_DRIVER,
'netapp_storage_family': app_constants.NETAPP_STORAGE_FAMILY,
@@ -414,17 +405,17 @@ class CinderHelm(openstack.OpenstackBaseHelm):
}
nfs_config = {
'netapp_storage_protocol': 'nfs',
'netapp_storage_protocol': app_constants.NETAPP_NFS_OPENSTACK_PROTOCOL,
'nfs_shares_config': app_constants.NFS_SHARES_CONFIG,
'nfs_mount_options': app_constants.NFS_MOUNT_OPTIONS,
}
iscsi_config = {
'netapp_storage_protocol': 'iscsi',
'netapp_storage_protocol': app_constants.NETAPP_ISCSI_OPENSTACK_PROTOCOL,
}
fc_config = {
'netapp_storage_protocol': 'fc',
'netapp_storage_protocol': app_constants.NETAPP_FC_OPENSTACK_PROTOCOL,
}
for backend in netapp_array:
python3-k8sapp-openstack/k8sapp_openstack/k8sapp_openstack/lifecycle/lifecycle_openstack.py
+12 -4
View File
@@ -437,10 +437,18 @@ class OpenstackAppLifecycleOperator(base.AppLifecycleOperator):
ceph_type=constants.SB_TYPE_CEPH_ROOK
)
netapp_backends_available = app_utils.check_netapp_backends()
netapp_nfs_available = netapp_backends_available.get("nfs", False)
netapp_iscsi_available = netapp_backends_available.get("iscsi", False)
netapp_fc_available = netapp_backends_available.get("fc", False)
netapp_nfs_available = netapp_backends_available.get(
app_constants.NETAPP_NFS_BACKEND_NAME,
False
)
netapp_iscsi_available = netapp_backends_available.get(
app_constants.NETAPP_ISCSI_BACKEND_NAME,
False
)
netapp_fc_available = netapp_backends_available.get(
app_constants.NETAPP_FC_BACKEND_NAME,
False
)
status = f"ceph_available={ceph_available}, " \
f"rook_ceph_available={rook_ceph_available}, " \
f"netapp_nfs_available={netapp_nfs_available}, " \
python3-k8sapp-openstack/k8sapp_openstack/k8sapp_openstack/tests/helm/test_cinder.py
+36 -12
View File
@@ -32,11 +32,15 @@ class CinderGetOverrideTest(CinderConversionTestCase,
)
@mock.patch(
'k8sapp_openstack.utils.check_netapp_backends',
return_value={'nfs': False, 'iscsi': False, 'fc': False}
return_value={app_constants.NETAPP_NFS_BACKEND_NAME: False,
app_constants.NETAPP_ISCSI_BACKEND_NAME: False,
app_constants.NETAPP_FC_BACKEND_NAME: False}
)
@mock.patch(
'k8sapp_openstack.helm.cinder.check_netapp_backends',
return_value={'nfs': False, 'iscsi': False, 'fc': False}
return_value={app_constants.NETAPP_NFS_BACKEND_NAME: False,
app_constants.NETAPP_ISCSI_BACKEND_NAME: False,
app_constants.NETAPP_FC_BACKEND_NAME: False}
)
@mock.patch(
'k8sapp_openstack.helm.cinder.is_user_overrides_available',
@@ -113,11 +117,15 @@ class CinderGetOverrideTest(CinderConversionTestCase,
)
@mock.patch(
'k8sapp_openstack.utils.check_netapp_backends',
return_value={'nfs': False, 'iscsi': False, 'fc': False}
return_value={app_constants.NETAPP_NFS_BACKEND_NAME: False,
app_constants.NETAPP_ISCSI_BACKEND_NAME: False,
app_constants.NETAPP_FC_BACKEND_NAME: False}
)
@mock.patch(
'k8sapp_openstack.helm.cinder.check_netapp_backends',
return_value={'nfs': False, 'iscsi': False, 'fc': False}
return_value={app_constants.NETAPP_NFS_BACKEND_NAME: False,
app_constants.NETAPP_ISCSI_BACKEND_NAME: False,
app_constants.NETAPP_FC_BACKEND_NAME: False}
)
@mock.patch(
'k8sapp_openstack.helm.cinder.is_user_overrides_available',
@@ -220,11 +228,15 @@ class CinderGetOverrideTest(CinderConversionTestCase,
)
@mock.patch(
'k8sapp_openstack.utils.check_netapp_backends',
return_value={'nfs': False, 'iscsi': False, 'fc': False}
return_value={app_constants.NETAPP_NFS_BACKEND_NAME: False,
app_constants.NETAPP_ISCSI_BACKEND_NAME: False,
app_constants.NETAPP_FC_BACKEND_NAME: False}
)
@mock.patch(
'k8sapp_openstack.helm.cinder.check_netapp_backends',
return_value={'nfs': False, 'iscsi': False, 'fc': False}
return_value={app_constants.NETAPP_NFS_BACKEND_NAME: False,
app_constants.NETAPP_ISCSI_BACKEND_NAME: False,
app_constants.NETAPP_FC_BACKEND_NAME: False}
)
@mock.patch(
'k8sapp_openstack.helm.cinder.is_user_overrides_available',
@@ -281,11 +293,15 @@ class CinderGetOverrideTest(CinderConversionTestCase,
)
@mock.patch(
'k8sapp_openstack.utils.check_netapp_backends',
return_value={'nfs': True, 'iscsi': False, 'fc': False}
return_value={app_constants.NETAPP_NFS_BACKEND_NAME: True,
app_constants.NETAPP_ISCSI_BACKEND_NAME: False,
app_constants.NETAPP_FC_BACKEND_NAME: False}
)
@mock.patch(
'k8sapp_openstack.helm.cinder.check_netapp_backends',
return_value={'nfs': True, 'iscsi': False, 'fc': False}
return_value={app_constants.NETAPP_NFS_BACKEND_NAME: True,
app_constants.NETAPP_ISCSI_BACKEND_NAME: False,
app_constants.NETAPP_FC_BACKEND_NAME: False}
)
@mock.patch(
'k8sapp_openstack.helm.cinder.is_user_overrides_available',
@@ -350,11 +366,15 @@ class CinderGetOverrideTest(CinderConversionTestCase,
)
@mock.patch(
'k8sapp_openstack.utils.check_netapp_backends',
return_value={'nfs': True, 'iscsi': False, 'fc': False}
return_value={app_constants.NETAPP_NFS_BACKEND_NAME: True,
app_constants.NETAPP_ISCSI_BACKEND_NAME: False,
app_constants.NETAPP_FC_BACKEND_NAME: False}
)
@mock.patch(
'k8sapp_openstack.helm.cinder.check_netapp_backends',
return_value={'nfs': True, 'iscsi': False, 'fc': False}
return_value={app_constants.NETAPP_NFS_BACKEND_NAME: True,
app_constants.NETAPP_ISCSI_BACKEND_NAME: False,
app_constants.NETAPP_FC_BACKEND_NAME: False}
)
@mock.patch(
'k8sapp_openstack.helm.cinder.is_user_overrides_available',
@@ -420,11 +440,15 @@ class CinderGetOverrideTest(CinderConversionTestCase,
)
@mock.patch(
'k8sapp_openstack.utils.check_netapp_backends',
return_value={'nfs': True, 'iscsi': True, 'fc': True}
return_value={app_constants.NETAPP_NFS_BACKEND_NAME: True,
app_constants.NETAPP_ISCSI_BACKEND_NAME: True,
app_constants.NETAPP_FC_BACKEND_NAME: True}
)
@mock.patch(
'k8sapp_openstack.helm.cinder.check_netapp_backends',
return_value={'nfs': True, 'iscsi': True, 'fc': True}
return_value={app_constants.NETAPP_NFS_BACKEND_NAME: True,
app_constants.NETAPP_ISCSI_BACKEND_NAME: True,
app_constants.NETAPP_FC_BACKEND_NAME: True}
)
@mock.patch(
'k8sapp_openstack.helm.cinder.is_user_overrides_available',
python3-k8sapp-openstack/k8sapp_openstack/k8sapp_openstack/tests/lifecycle/test_lifecycle_openstack.py
+15 -5
View File
@@ -34,7 +34,9 @@ class OpenstackAppLifecycleOperatorTest(dbbase.BaseHostTestCase):
return ceph_type == constants.SB_TYPE_CEPH, ""
@mock.patch('k8sapp_openstack.utils.check_netapp_backends',
return_value={"nfs": False, "iscsi": False, "fc": False})
return_value={app_constants.NETAPP_NFS_BACKEND_NAME: False,
app_constants.NETAPP_ISCSI_BACKEND_NAME: False,
app_constants.NETAPP_FC_BACKEND_NAME: False})
@mock.patch('k8sapp_openstack.utils.is_rook_ceph_api_available',
return_value=True)
@mock.patch('k8sapp_openstack.utils.get_ceph_fsid',
@@ -59,7 +61,9 @@ class OpenstackAppLifecycleOperatorTest(dbbase.BaseHostTestCase):
mock_is_rook_ceph_api_available.assert_called()
@mock.patch('k8sapp_openstack.utils.check_netapp_backends',
return_value={"nfs": False, "iscsi": False, "fc": False})
return_value={app_constants.NETAPP_NFS_BACKEND_NAME: False,
app_constants.NETAPP_ISCSI_BACKEND_NAME: False,
app_constants.NETAPP_FC_BACKEND_NAME: False})
@mock.patch('k8sapp_openstack.utils.get_ceph_fsid',
return_value='aa8c8da0-47de-4fad-8b5d-2c06be236fc8')
@mock.patch('k8sapp_openstack.utils.is_ceph_backend_available')
@@ -80,7 +84,9 @@ class OpenstackAppLifecycleOperatorTest(dbbase.BaseHostTestCase):
mock_get_ceph_fsid.assert_called()
@mock.patch('k8sapp_openstack.utils.check_netapp_backends',
return_value={"nfs": True, "iscsi": False, "fc": False})
return_value={app_constants.NETAPP_NFS_BACKEND_NAME: True,
app_constants.NETAPP_ISCSI_BACKEND_NAME: False,
app_constants.NETAPP_FC_BACKEND_NAME: False})
@mock.patch('k8sapp_openstack.utils.get_ceph_fsid', return_value=None)
@mock.patch('k8sapp_openstack.utils.is_ceph_backend_available')
def test_semantic_check_storage_backend_available_netapp_nfs(
@@ -100,7 +106,9 @@ class OpenstackAppLifecycleOperatorTest(dbbase.BaseHostTestCase):
mock_get_ceph_fsid.assert_called()
@mock.patch('k8sapp_openstack.utils.check_netapp_backends',
return_value={"nfs": False, "iscsi": False, "fc": False})
return_value={app_constants.NETAPP_NFS_BACKEND_NAME: False,
app_constants.NETAPP_ISCSI_BACKEND_NAME: False,
app_constants.NETAPP_FC_BACKEND_NAME: False})
@mock.patch('k8sapp_openstack.utils.get_ceph_fsid', return_value=None)
@mock.patch('k8sapp_openstack.utils.is_ceph_backend_available')
def test_semantic_check_storage_backend_available_fsid_unavailable(
@@ -124,7 +132,9 @@ class OpenstackAppLifecycleOperatorTest(dbbase.BaseHostTestCase):
mock_check_netapp_backends.assert_called()
@mock.patch('k8sapp_openstack.utils.check_netapp_backends',
return_value={"nfs": False, "iscsi": False, "fc": False})
return_value={app_constants.NETAPP_NFS_BACKEND_NAME: False,
app_constants.NETAPP_ISCSI_BACKEND_NAME: False,
app_constants.NETAPP_FC_BACKEND_NAME: False})
@mock.patch('k8sapp_openstack.utils.get_ceph_fsid', return_value=None)
@mock.patch('k8sapp_openstack.utils.is_ceph_backend_available',
side_effect=[(False, ""), (False, "")])
python3-k8sapp-openstack/k8sapp_openstack/k8sapp_openstack/tests/utils/test_utils.py
+271 -119
View File
@@ -1485,109 +1485,78 @@ class UtilsTest(dbbase.ControllerHostTestCase):
assert result == ""
@mock.patch('k8sapp_openstack.utils.send_cmd_read_response',
return_value="ontap-nas:nfs:iscsi")
@mock.patch('k8sapp_openstack.utils.is_netapp_storageclass_available',
return_value=True)
@mock.patch('k8sapp_openstack.utils._get_value_from_application',
return_value='trident')
@mock.patch('k8sapp_openstack.utils.is_user_overrides_available',
@mock.patch("k8sapp_openstack.utils.is_user_overrides_available",
return_value=False)
@mock.patch('sysinv.common.kubernetes.KubeOperator')
def test_check_netapp_backends_nfs(self, mock_kube_operator, *_):
""" Test if check_netapp_backends can find the 'nfs' backend
@mock.patch("k8sapp_openstack.utils.netapp_backends_auto_discovery",
return_value={
app_constants.NETAPP_NFS_BACKEND_NAME: True,
app_constants.NETAPP_ISCSI_BACKEND_NAME: True,
app_constants.NETAPP_FC_BACKEND_NAME: False,
})
def test_check_netapp_backends_no_overrides(self, *_):
"""Test when user overrides are not available
(return discovery results as-is).
"""
# Mocks for Netapp pods checking
mock_pod_list = [mock.MagicMock()]
kube_operator_instance = mock_kube_operator.return_value
kube_operator_instance.kube_get_pods_by_selector.return_value = mock_pod_list
result = app_utils.check_netapp_backends()
self.assertEqual(
result,
{
app_constants.NETAPP_NFS_BACKEND_NAME: True,
app_constants.NETAPP_ISCSI_BACKEND_NAME: True,
app_constants.NETAPP_FC_BACKEND_NAME: False,
}
)
backends_map = app_utils.check_netapp_backends()
assert backends_map["nfs"]
@mock.patch('k8sapp_openstack.utils.send_cmd_read_response',
return_value="ontap-san:nfs:iscsi")
@mock.patch('k8sapp_openstack.utils.is_netapp_storageclass_available',
@mock.patch("k8sapp_openstack.utils.get_enabled_storage_backends_from_override",
return_value=[
app_constants.NETAPP_NFS_BACKEND_NAME,
app_constants.NETAPP_FC_BACKEND_NAME,
])
@mock.patch("k8sapp_openstack.utils.is_user_overrides_available",
return_value=True)
@mock.patch('k8sapp_openstack.utils._get_value_from_application',
return_value='trident')
@mock.patch('k8sapp_openstack.utils.is_user_overrides_available',
return_value=False)
@mock.patch('sysinv.common.kubernetes.KubeOperator')
def test_check_netapp_backends_iscsi(self, mock_kube_operator, *_):
""" Test if check_netapp_backends can find the 'iscsi' backend
"""
mock_pod_list = [mock.MagicMock()]
kube_operator_instance = mock_kube_operator.return_value
kube_operator_instance.kube_get_pods_by_selector.return_value = mock_pod_list
@mock.patch("k8sapp_openstack.utils.netapp_backends_auto_discovery",
return_value={
app_constants.NETAPP_NFS_BACKEND_NAME: True,
app_constants.NETAPP_ISCSI_BACKEND_NAME: True,
app_constants.NETAPP_FC_BACKEND_NAME: True,
})
def test_check_netapp_backends_enabled_subset(self, *_):
"""Test overrides restrict to a subset of discovered backends."""
result = app_utils.check_netapp_backends()
self.assertEqual(
result,
{
app_constants.NETAPP_NFS_BACKEND_NAME: True,
app_constants.NETAPP_ISCSI_BACKEND_NAME: False,
app_constants.NETAPP_FC_BACKEND_NAME: True,
}
)
backends_map = app_utils.check_netapp_backends()
assert backends_map["iscsi"]
@mock.patch('k8sapp_openstack.utils.send_cmd_read_response',
return_value="ontap-san:nfs:fc")
@mock.patch('k8sapp_openstack.utils.is_netapp_storageclass_available',
@mock.patch("k8sapp_openstack.utils.get_enabled_storage_backends_from_override",
return_value=[
app_constants.NETAPP_NFS_BACKEND_NAME,
app_constants.NETAPP_ISCSI_BACKEND_NAME,
app_constants.NETAPP_FC_BACKEND_NAME,
])
@mock.patch("k8sapp_openstack.utils.is_user_overrides_available",
return_value=True)
@mock.patch('k8sapp_openstack.utils._get_value_from_application',
return_value='trident')
@mock.patch('k8sapp_openstack.utils.is_user_overrides_available',
return_value=False)
@mock.patch('sysinv.common.kubernetes.KubeOperator')
def test_check_netapp_backends_fc(self, mock_kube_operator, *_):
""" Test if check_netapp_backends can find the 'fc' backend
"""
mock_pod_list = [mock.MagicMock()]
kube_operator_instance = mock_kube_operator.return_value
kube_operator_instance.kube_get_pods_by_selector.return_value = mock_pod_list
backends_map = app_utils.check_netapp_backends()
assert backends_map["fc"]
@mock.patch('k8sapp_openstack.utils.send_cmd_read_response',
return_value="ontap-san:nfs:iscsi\nontap-nas:nfs:iscsi")
@mock.patch('k8sapp_openstack.utils.is_netapp_storageclass_available',
return_value=True)
@mock.patch('k8sapp_openstack.utils._get_value_from_application',
return_value='trident')
@mock.patch('k8sapp_openstack.utils.is_user_overrides_available',
return_value=False)
@mock.patch('sysinv.common.kubernetes.KubeOperator')
def test_check_netapp_backends_multiple(self, mock_kube_operator, *_):
""" Test if check_netapp_backends can find multiple backends
"""
mock_pod_list = [mock.MagicMock()]
kube_operator_instance = mock_kube_operator.return_value
kube_operator_instance.kube_get_pods_by_selector.return_value = mock_pod_list
backends_map = app_utils.check_netapp_backends()
assert backends_map["nfs"] and backends_map["iscsi"]
@mock.patch('k8sapp_openstack.utils.send_cmd_read_response',
return_value="")
@mock.patch('k8sapp_openstack.utils.is_netapp_storageclass_available',
return_value=True)
@mock.patch('k8sapp_openstack.utils._get_value_from_application',
return_value='trident')
@mock.patch('k8sapp_openstack.utils.get_enabled_storage_backends_from_override',
return_value=["ceph"])
@mock.patch('k8sapp_openstack.utils.is_user_overrides_available',
return_value=False)
@mock.patch('sysinv.common.kubernetes.KubeOperator')
def test_check_netapp_backends_none(self, mock_kube_operator, *_):
""" Test if check_netapp_backends returns a backend_map with all
values set to false when no backends are available
"""
mock_pod_list = [mock.MagicMock()]
kube_operator_instance = mock_kube_operator.return_value
kube_operator_instance.kube_get_pods_by_selector.return_value = mock_pod_list
backends_map = app_utils.check_netapp_backends()
assert not any(backends_map.values())
@mock.patch("k8sapp_openstack.utils.netapp_backends_auto_discovery",
return_value={
app_constants.NETAPP_NFS_BACKEND_NAME: True,
app_constants.NETAPP_ISCSI_BACKEND_NAME: True,
app_constants.NETAPP_FC_BACKEND_NAME: False,
})
def test_check_netapp_backends_discovery_limits(self, *_):
"""Test discovery availability still limits enabled backends."""
result = app_utils.check_netapp_backends()
self.assertEqual(
result,
{
app_constants.NETAPP_NFS_BACKEND_NAME: True,
app_constants.NETAPP_ISCSI_BACKEND_NAME: True,
app_constants.NETAPP_FC_BACKEND_NAME: False,
}
)
@mock.patch('k8sapp_openstack.utils._get_value_from_application')
def test_get_enabled_storage_backends_from_override_netapp_nfs(self, mock_get_values):
@@ -1646,25 +1615,6 @@ class UtilsTest(dbbase.ControllerHostTestCase):
priority_list = app_utils.get_storage_backends_priority_list(chart)
assert priority_list == ["netapp-nfs", "ceph", "netapp-iscsi", "netapp-fc"]
@mock.patch('k8sapp_openstack.utils.is_netapp_storageclass_available', return_value=False)
@mock.patch('k8sapp_openstack.utils._get_value_from_application', return_value='trident')
@mock.patch('k8sapp_openstack.utils.is_user_overrides_available', return_value=True)
@mock.patch('k8sapp_openstack.utils.get_enabled_storage_backends_from_override')
@mock.patch('sysinv.common.kubernetes.KubeOperator')
def test_check_netapp_backends_from_override(self, mock_kube_operator, mock_enabled_backends, *_):
""" Test if check_netapp_backends returns a backend_map with the
values present in the override file
"""
mock_pod_list = [mock.MagicMock()]
kube_operator_instance = mock_kube_operator.return_value
kube_operator_instance.kube_get_pods_by_selector.return_value = mock_pod_list
mock_enabled_backends.return_value = ["netapp-nfs"]
backends_map = app_utils.check_netapp_backends()
assert backends_map["nfs"]
@mock.patch("k8sapp_openstack.utils.send_cmd_read_response",
return_value="netapp-nas-backend other-nas")
def test_get_netapp_storage_class_name_nfs(self, *_):
@@ -1724,7 +1674,9 @@ class UtilsTest(dbbase.ControllerHostTestCase):
assert result == "example.com"
@mock.patch("k8sapp_openstack.utils.check_netapp_backends",
return_value={"nfs": True, "iscsi": True, "fc": False})
return_value={app_constants.NETAPP_NFS_BACKEND_NAME: True,
app_constants.NETAPP_ISCSI_BACKEND_NAME: True,
app_constants.NETAPP_FC_BACKEND_NAME: False})
@mock.patch("k8sapp_openstack.utils.get_ceph_rbd_storage_class_name",
return_value="rook-ceph-rbd")
@mock.patch("k8sapp_openstack.utils.get_netapp_storage_class_name")
@@ -1766,7 +1718,9 @@ class UtilsTest(dbbase.ControllerHostTestCase):
)
@mock.patch("k8sapp_openstack.utils.check_netapp_backends",
return_value={"nfs": True, "iscsi": False, "fc": True})
return_value={app_constants.NETAPP_NFS_BACKEND_NAME: True,
app_constants.NETAPP_ISCSI_BACKEND_NAME: False,
app_constants.NETAPP_FC_BACKEND_NAME: True})
@mock.patch("k8sapp_openstack.utils.get_ceph_rbd_storage_class_name",
return_value="ignored")
@mock.patch("k8sapp_openstack.utils.get_netapp_storage_class_name")
@@ -1808,7 +1762,9 @@ class UtilsTest(dbbase.ControllerHostTestCase):
)
@mock.patch("k8sapp_openstack.utils.check_netapp_backends",
return_value={"nfs": False, "iscsi": False, "fc": False})
return_value={app_constants.NETAPP_NFS_BACKEND_NAME: False,
app_constants.NETAPP_ISCSI_BACKEND_NAME: False,
app_constants.NETAPP_FC_BACKEND_NAME: False})
@mock.patch("k8sapp_openstack.utils.get_ceph_rbd_storage_class_name",
return_value="ceph-rbd")
@mock.patch("k8sapp_openstack.utils.get_netapp_storage_class_name")
@@ -1936,3 +1892,199 @@ class UtilsTest(dbbase.ControllerHostTestCase):
result = app_utils.get_ceph_rbd_storage_class_name()
self.assertEqual(result, "")
mock_get_sc.assert_called_once_with(app_constants.CEPH_ROOK_RBD_DRIVER)
@mock.patch("k8sapp_openstack.utils.send_cmd_read_response")
def test_discover_netapp_credentials_nfs(self, mock_send, *_):
"""Test discovery for netapp-nfs using ontap-nas driver."""
# First call → TBC Secret name; second → "username::password" (b64)
mock_send.side_effect = [
"tbc-nas-secret",
"dXNlcg==::cHdk" # base64(user)::base64(pwd)
]
result = app_utils.discover_netapp_credentials(
app_constants.NETAPP_NFS_BACKEND_NAME
)
self.assertEqual(
result,
{"netapp_login": "user", "netapp_password": "pwd"}
)
@mock.patch("k8sapp_openstack.utils.send_cmd_read_response",
return_value="")
def test_discover_netapp_credentials_empty(self, *_):
"""Test when TBC exposes no Secret or it cannot be found."""
result = app_utils.discover_netapp_credentials(
app_constants.NETAPP_NFS_BACKEND_NAME
)
self.assertEqual(
result,
{}
)
@mock.patch("k8sapp_openstack.utils.send_cmd_read_response",
return_value="10.0.0.10:::svm_iscsi")
def test_discover_netapp_configs_iscsi(self, *_):
"""Test discovery for netapp-iscsi (ontap-san)."""
result = app_utils.discover_netapp_configs(
app_constants.NETAPP_ISCSI_BACKEND_NAME
)
self.assertEqual(
result,
{
"volume_driver": app_constants.NETAPP_CINDER_VOLUME_DRIVER,
"netapp_storage_family": app_constants.NETAPP_STORAGE_FAMILY,
"netapp_storage_protocol": app_constants.NETAPP_BACKEND_TO_OPENSTACK_PROTOCOL[
app_constants.NETAPP_ISCSI_BACKEND_NAME
],
"netapp_vserver": "svm_iscsi",
"netapp_server_hostname": "10.0.0.10",
"netapp_server_port": app_constants.NETAPP_DEFAULT_SERVER_PORT,
"netapp_transport_type": (
app_constants.NETAPP_DEFAULT_SERVER_TRANSPORT_TYPE
),
}
)
@mock.patch("k8sapp_openstack.utils.send_cmd_read_response",
return_value="10.0.0.20:::svm_nfs")
def test_discover_netapp_configs_nfs(self, *_):
"""Test discovery for netapp-nfs (ontap-nas)."""
result = app_utils.discover_netapp_configs(
app_constants.NETAPP_NFS_BACKEND_NAME
)
self.assertEqual(
result,
{
"volume_driver": app_constants.NETAPP_CINDER_VOLUME_DRIVER,
"netapp_storage_family": app_constants.NETAPP_STORAGE_FAMILY,
"netapp_storage_protocol": app_constants.NETAPP_BACKEND_TO_OPENSTACK_PROTOCOL[
app_constants.NETAPP_NFS_BACKEND_NAME
],
"netapp_vserver": "svm_nfs",
"netapp_server_hostname": "10.0.0.20",
"netapp_server_port": app_constants.NETAPP_DEFAULT_SERVER_PORT,
"netapp_transport_type": (
app_constants.NETAPP_DEFAULT_SERVER_TRANSPORT_TYPE
),
}
)
@mock.patch("k8sapp_openstack.utils.send_cmd_read_response",
return_value="")
def test_discover_netapp_configs_empty(self, *_):
"""Test when managementLIF:::svm cannot be retrieved."""
result = app_utils.discover_netapp_configs(
app_constants.NETAPP_NFS_BACKEND_NAME
)
self.assertEqual(result, {})
@mock.patch("k8sapp_openstack.utils.send_cmd_read_response",
side_effect=RuntimeError("kubectl failed"))
def test_discover_netapp_configs_cmd_exception(self, *_):
"""Test when command execution raises an exception."""
result = app_utils.discover_netapp_configs(
app_constants.NETAPP_ISCSI_BACKEND_NAME
)
self.assertEqual(result, {})
def test_override_configs_scalar(self):
"""Test overriding scalar values and adding new keys."""
original = {
'db': {'host': 'localhost', 'port': 3306},
'debug': False
}
overrides = {
'db': {'port': 5432},
'debug': True,
'new_key': 'value'
}
result = app_utils.override_configs(original, overrides)
self.assertEqual(
result,
{
'db': {'host': 'localhost', 'port': 5432},
'debug': True,
'new_key': 'value'
}
)
def test_override_configs_nested_merge(self):
"""Test recursive merge for nested dictionaries."""
original = {
'database': {
'host': 'localhost',
'port': 3306,
'credentials': {
'username': 'admin', 'password': 'admin123'
}
}
}
overrides = {
'database': {
'port': 5432,
'credentials': {'password': 'newpassword'}
}
}
result = app_utils.override_configs(original, overrides)
self.assertEqual(
result,
{
'database': {
'host': 'localhost',
'port': 5432,
'credentials': {
'username': 'admin',
'password': 'newpassword'
}
}
}
)
def test_override_configs_list_replacement(self):
"""Test lists are replaced, not merged."""
original = {
'features': ['feature1', 'feature2'],
'flags': [1, 2, 3]
}
overrides = {
'features': ['feature3'],
'flags': []
}
result = app_utils.override_configs(original, overrides)
self.assertEqual(
result,
{
'features': ['feature3'],
'flags': []
}
)
def test_override_configs_original_immutability(self):
"""Test original dict is not modified (deep copy behavior)."""
original = {
'section': {
'sub': {'a': 1, 'b': 2}
},
'list': [1, 2]
}
overrides = {
'section': {'sub': {'b': 20, 'c': 3}},
'list': [99]
}
# Keep a snapshot of original before calling
original_snapshot = {
'section': {'sub': {'a': 1, 'b': 2}},
'list': [1, 2]
}
result = app_utils.override_configs(original, overrides)
# Result reflects overrides
self.assertEqual(
result,
{
'section': {'sub': {'a': 1, 'b': 20, 'c': 3}},
'list': [99]
}
)
# Original remains unchanged
self.assertEqual(original, original_snapshot)
python3-k8sapp-openstack/k8sapp_openstack/k8sapp_openstack/utils.py
+316 -30
View File
@@ -3,6 +3,7 @@
#
# SPDX-License-Identifier: Apache-2.0
#
from copy import deepcopy
from grp import getgrnam
import json
import os
@@ -101,6 +102,68 @@ def is_user_overrides_available(chart_name, override_name) -> bool:
return override is not None
def override_configs(config: dict, overrides: dict) -> dict:
""" Override configuration dictionary with another one.
Args:
config (dict): The original configuration dictionary.
overrides (dict): The dictionary with overriding values.
Returns:
dict: The resulting configuration dictionary after applying overrides.
Note:
- This function performs a deep copy of the original configuration to avoid
modifying it in place.
- If both the original and overriding values for a key are dictionaries,
the function merges them recursively.
- For lists and other non-dictionary types, the overriding value completely
replaces the original value.
Example:
original_config = {
'database': {
'host': 'localhost',
'port': 3306,
'credentials': {
'username': 'admin',
'password': 'admin123'
}
},
'features': ['feature1', 'feature2']
}
overrides = {
'database': {
'port': 5432,
'credentials': {
'password': 'newpassword'
}
},
'features': ['feature3']
}
result = override_configs(original_config, overrides)
# result will be:
{
'database': {
'host': 'localhost',
'port': 5432,
'credentials': {
'username': 'admin',
'password': 'newpassword'
}
},
'features': ['feature3']
}
"""
result = deepcopy(config)
for k, val in overrides.items():
if k in result and isinstance(result[k], dict) and isinstance(val, dict):
result[k] = override_configs(result[k], val)
else:
result[k] = deepcopy(val)
return result
def get_services_fqdn_pattern() -> str:
"""Get services FQDN configuration pattern
@@ -578,25 +641,228 @@ def is_netapp_storageclass_available() -> bool:
return bool(netapp_storageclasses)
def check_netapp_backends() -> dict:
def check_netapp_backends(chart_name: str = app_constants.HELM_CHART_CINDER,
override_name: str = app_constants.OVERRIDE_STORAGE_BACKENDS) -> dict:
"""
Check for the presence of NetApp backends (NFS, iSCSI and SCSi via FC) using
either the user overrides or the auto-discovery feature
Check the availability of NetApp backends, considering user overrides.
To be considered a available backend, two conditions must be met:
1. The backend must be discovered automatically via `netapp_backends_auto_discovery()`.
2. If user overrides are defined for storage backends, the backend must be
explicitly enabled in those overrides.
Args:
chart_name (str): The Helm chart name to check for user overrides.
Defaults to `app_constants.HELM_CHART_CINDER`.
override_name (str): The name of the override field in values.yaml.
Returns:
dict: A dictionary indicating the availability of 'nfs', 'iscsi' or 'fc' backends.
Example: {"nfs": True, "iscsi": False, "fc": False}
dict: A dictionary indicating the availability of each NetApp backend:
```
{
"netapp-nfs": True/False,
"netapp-iscsi": True/False,
"netapp-fc": True/False
}
```
"""
if is_user_overrides_available(app_constants.HELM_CHART_CINDER,
app_constants.OVERRIDE_STORAGE_BACKENDS):
enabled_backends = get_enabled_storage_backends_from_override()
backends_map = {"nfs": False, "iscsi": False, "fc": False}
backends_map["nfs"] = app_constants.NETAPP_NFS_BACKEND_NAME in enabled_backends
backends_map["iscsi"] = app_constants.NETAPP_ISCSI_BACKEND_NAME in enabled_backends
backends_map["fc"] = app_constants.NETAPP_FC_BACKEND_NAME in enabled_backends
return backends_map
netapp_backends_available = netapp_backends_auto_discovery()
if is_user_overrides_available(chart_name=chart_name,
override_name=override_name):
enabled_backends = get_enabled_storage_backends_from_override(
chart_name=chart_name,
override_name=override_name
)
netapp_backends_available[app_constants.NETAPP_NFS_BACKEND_NAME] &= (
app_constants.NETAPP_NFS_BACKEND_NAME in enabled_backends
)
netapp_backends_available[app_constants.NETAPP_ISCSI_BACKEND_NAME] &= (
app_constants.NETAPP_ISCSI_BACKEND_NAME in enabled_backends
)
netapp_backends_available[app_constants.NETAPP_FC_BACKEND_NAME] &= (
app_constants.NETAPP_FC_BACKEND_NAME in enabled_backends
)
return netapp_backends_available
return netapp_backends_auto_discovery()
def discover_netapp_credentials(backend_type: str) -> dict:
"""
Discover NetApp backend credentials (username and password) from Kubernetes Secrets
referenced by TridentBackendConfig (TBC).
This function assumes that only one protocol is deployed for the requested
storage driver at a time. For example, `netapp-iscsi` and `netapp-fc` both use
``storageDriverName: ontap-san``, but you do **not** run iSCSI and FC simultaneously.
Likewise, `netapp-nfs` uses ``storageDriverName: ontap-nas``.
Args:
backend_type (str):
Logical backend type to query. Supported values:
- `"netapp-nfs"`: filters TBCs with ``storageDriverName == "ontap-nas"``.
- `"netapp-iscsi"`: filters TBCs with ``storageDriverName == "ontap-san"``.
- `"netapp-fc"`: filters TBCs with ``storageDriverName == "ontap-san"``.
Only **one** of the SAN protocols (iSCSI or FC) should be present at a time.
If multiple TBCs match, the **first** match is used.
Returns:
dict:
A dictionary with the decoded credentials:
```
{
"netapp_login": "<username>",
"netapp_password": "<password>"
}
```
If the Secret or fields cannot be found/decoded, an empty dict
will be returned
Example:
>>> discover_netapp_credentials("netapp-iscsi")
{"netapp_login": "<username>", "netapp_password": <password>"}
>>> discover_netapp_credentials("invalid-backend")
{}
"""
credentials = dict()
bt = backend_type.strip().lower()
if bt not in app_constants.NETAPP_SUPPORTED_BACKENDS:
LOG.error(
f"Unsupported backend '{bt}'. "
f"Supported backends: {app_constants.NETAPP_SUPPORTED_BACKENDS}"
)
return credentials
driver = app_constants.NETAPP_BACKEND_TO_TYPE[bt]
jsonpath = f"{{.items[?(@.spec.storageDriverName==\"{driver}\")].spec.credentials.name}}"
cmd = [
"kubectl", "--kubeconfig", kubernetes.KUBERNETES_ADMIN_CONF,
"get", "tridentbackendconfigs",
"-n", app_constants.OPENSTACK_NETAPP_NAMESPACE,
"-o", f"jsonpath={jsonpath}"
]
try:
# Discover credentials secret name via tridentbackendconfigs filtered by
# storageDriverName
secret_name = send_cmd_read_response(cmd, log=False)
if not secret_name:
LOG.error(f"No tridentbackendconfigs with storageDriverName='{driver}'"
"found or missing '.spec.credentials.name' definition.")
return credentials
secret_name = secret_name.splitlines()[0].split()[0] # First match
# Read base64(username::password) from the Secret via jsonpath
cmd = [
"kubectl", "--kubeconfig", kubernetes.KUBERNETES_ADMIN_CONF,
"get", "secret", secret_name,
"-n", app_constants.OPENSTACK_NETAPP_NAMESPACE,
"-o", "jsonpath={.data.username}{\"::\"}{.data.password}"
]
b64_creds = send_cmd_read_response(cmd, log=False)
if not b64_creds or "::" not in b64_creds:
LOG.error(f"Secret '{secret_name}' missing credentials")
return credentials
b64_user, b64_pass = b64_creds.split("::", 1)
# Decode base64 credentials
username = base64.decode_as_text(b64_user)
password = base64.decode_as_text(b64_pass)
credentials['netapp_login'] = username
credentials['netapp_password'] = password
except Exception as e:
LOG.error("Error recovering credentials for '{backend_type}' backend: "
f"{e}")
return credentials
def discover_netapp_configs(backend_type: str) -> dict:
"""
Discover OpenStack-compatible NetApp backend configuration from
TridentBackendConfig (TBC) objects in the designated namespace.
This function follows the same style and assumptions as
``discover_netapp_credentials``:
- Filters TBCs **only** by ``.spec.storageDriverName``.
- Assumes that for SAN drivers (``ontap-san``), only **one** protocol
is deployed at a time (either iSCSI or FC). Because of this, the
**first** matching TBC is considered authoritative.
- Uses a single `kubectl` jsonpath call that concatenates both
``managementLIF`` and ``svm`` with a sentinel delimiter, then splits
locally in Python.
Args:
backend_type (str):
Logical backend type to query. Supported values:
- `"netapp-nfs"`: filters TBCs with ``storageDriverName == "ontap-nas"``.
- `"netapp-iscsi"`: filters TBCs with ``storageDriverName == "ontap-san"``.
- `"netapp-fc"`: filters TBCs with ``storageDriverName == "ontap-san"``.
Only **one** of the SAN protocols (iSCSI or FC) should be present at a time.
If multiple TBCs match, the **first** match is used.
Returns:
dict:
An OpenStack-compatible configuration dict. If discovery fails (e.g.,
TBC fields missing), an **empty dict** is returned and an error is logged.
Returned object (OpenStack Cinder NetApp unified driver compatible):
{
"netapp_storage_protocol": "nfs" | "iscsi" | "fc",
"netapp_server_hostname": "<managementLIF>",
"netapp_server_port": <int>,
"netapp_transport_type": "http" | "https",
"netapp_vserver": "<svm>"
}
Example:
>>> discover_netapp_configs("netapp-iscsi")
{
"volume_driver": "cinder.volume.drivers.netapp.common.NetAppDriver",
"netapp_storage_family": "ontap_cluster",
"netapp_storage_protocol": "iscsi",
"netapp_vserver": "svm_iscsi",
"netapp_server_hostname": "10.0.0.10",
"netapp_server_port": 443,
"netapp_transport_type": "https",
}
"""
openstack_config = dict()
bt = backend_type.strip().lower()
if bt not in app_constants.NETAPP_SUPPORTED_BACKENDS:
LOG.error(
f"Unsupported backend '{bt}'. "
f"Supported backends: {app_constants.NETAPP_SUPPORTED_BACKENDS}"
)
return openstack_config
driver = app_constants.NETAPP_BACKEND_TO_TYPE[bt]
netapp_protocol = app_constants.NETAPP_BACKEND_TO_OPENSTACK_PROTOCOL[bt]
# Read managementLIF:::svm from tridentbackendconfigs via jsonpath
jsonpath = (
f"{{.items[?(@.spec.storageDriverName==\"{driver}\")].spec.managementLIF}}"
f"{{\":::\"}}"
f"{{.items[?(@.spec.storageDriverName==\"{driver}\")].spec.svm}}"
)
cmd = [
"kubectl", "--kubeconfig", kubernetes.KUBERNETES_ADMIN_CONF,
"get", "tridentbackendconfigs",
"-n", app_constants.OPENSTACK_NETAPP_NAMESPACE,
"-o", f"jsonpath={jsonpath}"
]
try:
mgmt_lif_svm = send_cmd_read_response(cmd, log=False).strip()
if not mgmt_lif_svm or ":::" not in mgmt_lif_svm:
LOG.error("Could not retrieve managementLIF and svm for "
f"driver='{driver}'.")
return openstack_config
mgmt_lif_svm = mgmt_lif_svm.splitlines()[0].split()[0] # First match
mgmt_lif, svm = mgmt_lif_svm.split(":::", 1)
if not mgmt_lif or not svm:
LOG.error(f"Missing managementLIF or svm for driver='{driver}'.")
return openstack_config
openstack_config["volume_driver"] = app_constants.NETAPP_CINDER_VOLUME_DRIVER
openstack_config["netapp_storage_family"] = app_constants.NETAPP_STORAGE_FAMILY
openstack_config["netapp_storage_protocol"] = netapp_protocol
openstack_config["netapp_vserver"] = svm
openstack_config["netapp_server_hostname"] = mgmt_lif
openstack_config["netapp_server_port"] = app_constants.NETAPP_DEFAULT_SERVER_PORT
openstack_config["netapp_transport_type"] = app_constants.NETAPP_DEFAULT_SERVER_TRANSPORT_TYPE
except Exception as e:
LOG.error(f"Error recovering configs for '{backend_type}' backend: {e}")
return openstack_config
def netapp_backends_auto_discovery() -> dict:
@@ -609,14 +875,18 @@ def netapp_backends_auto_discovery() -> dict:
Returns:
dict: A dictionary indicating the availability of 'nfs', 'iscsi' or 'fc' backends.
Example: {"nfs": True, "iscsi": False, "fc": False}
Example: {"netapp-nfs": True, "netapp-iscsi": False, "netapp-fc": False}
"""
namespace = _get_value_from_application(
default_value=app_constants.OPENSTACK_NETAPP_NAMESPACE,
chart_name=app_constants.HELM_CHART_CLIENTS,
override_name="netAppNamespace")
backends_map = {"nfs": False, "iscsi": False, "fc": False}
backends_map = {
app_constants.NETAPP_NFS_BACKEND_NAME: False,
app_constants.NETAPP_ISCSI_BACKEND_NAME: False,
app_constants.NETAPP_FC_BACKEND_NAME: False
}
try:
kube = kubernetes.KubeOperator()
@@ -629,10 +899,6 @@ def netapp_backends_auto_discovery() -> dict:
f" with 'app={app_constants.NETAPP_CONTROLLER_LABEL}' label")
return backends_map
# Checking if we have NetApp storageclasses
if not is_netapp_storageclass_available():
return backends_map
# Searching for available NetApp backend protocols
# The output will be in the format:
# <Storage Drive Name>:<NAS Type>:<SAN Type>
@@ -662,8 +928,21 @@ def netapp_backends_auto_discovery() -> dict:
protocols_found = []
for info in protocol_info.split("\n"):
if "ontap-nas" in info:
protocols_found.append(info.split(":")[1])
nas_type = info.split(":")[1].strip()
if nas_type == app_constants.NETAPP_NFS_NAS_TYPE:
protocols_found.append(app_constants.NETAPP_NFS_BACKEND_NAME)
else:
LOG.warning(f"Unknown NAS type '{nas_type}' found in "
"trident backends")
elif "ontap-san" in info:
san_type = info.split(":")[2].strip()
if san_type == app_constants.NETAPP_FC_SAN_TYPE:
protocols_found.append(app_constants.NETAPP_FC_BACKEND_NAME)
elif san_type == app_constants.NETAPP_ISCSI_SAN_TYPE:
protocols_found.append(app_constants.NETAPP_ISCSI_BACKEND_NAME)
else:
LOG.warning(f"Unknown SAN type '{san_type}' found in "
"trident backends")
protocols_found.append(info.split(":")[2])
# Updating backends_map
@@ -1023,11 +1302,18 @@ def get_vswitch_label_from_override_file() -> set:
return set(labels)
def get_enabled_storage_backends_from_override() -> list:
def get_enabled_storage_backends_from_override(
chart_name: str = app_constants.HELM_CHART_CINDER,
override_name: str = app_constants.OVERRIDE_STORAGE_BACKENDS
) -> list:
"""
Retrieves the available storage backends from Cinder's
override file
Args:
chart_name (str): The Helm Chart name
override_name (str): The name of the override field in values.yaml.
Returns:
list: a list of enabled backends names (for instance,
["ceph", "netap_nfs"]), as defined in the overrides file
@@ -1035,8 +1321,8 @@ def get_enabled_storage_backends_from_override() -> list:
storage_backends = _get_value_from_application(
default_value=app_constants.DEFAULT_STORAGE_BACKEND_SELECT,
chart_name=app_constants.HELM_CHART_CINDER,
override_name=app_constants.OVERRIDE_STORAGE_BACKENDS
chart_name=chart_name,
override_name=override_name
)
if storage_backends:
@@ -1740,19 +2026,19 @@ def get_available_volume_backends() -> dict:
netapp_backend = check_netapp_backends()
available_volume_backends = {
"ceph": ceph_storage_class,
"netapp-nfs": (
app_constants.NETAPP_NFS_BACKEND_NAME: (
get_netapp_storage_class_name(app_constants.BACKEND_TYPE_NETAPP_NFS)
if netapp_backend.get("nfs", False)
if netapp_backend.get(app_constants.NETAPP_NFS_BACKEND_NAME, False)
else ""
),
"netapp-iscsi": (
app_constants.NETAPP_ISCSI_BACKEND_NAME: (
get_netapp_storage_class_name(app_constants.BACKEND_TYPE_NETAPP_ISCSI)
if netapp_backend.get("iscsi", False)
if netapp_backend.get(app_constants.NETAPP_ISCSI_BACKEND_NAME, False)
else ""
),
"netapp-fc": (
app_constants.NETAPP_FC_BACKEND_NAME: (
get_netapp_storage_class_name(app_constants.BACKEND_TYPE_NETAPP_FC)
if netapp_backend.get("fc", False)
if netapp_backend.get(app_constants.NETAPP_FC_BACKEND_NAME, False)
else ""
),
}