From 87c00d79a6a8be82e4513b5679e6da677d504629 Mon Sep 17 00:00:00 2001 From: Uri Ziv Date: Thu, 29 Jan 2026 13:37:27 +0200 Subject: [PATCH] CSI-5997 --- common/config.yaml | 3 - controllers/common/node_info.py | 13 +- controllers/common/settings.py | 6 +- controllers/common/utils.py | 19 --- .../servers/csi/csi_controller_server.py | 4 +- controllers/servers/csi/requirements.txt | 1 + controllers/servers/host_definer/hd_types.py | 14 ++- .../host_definer/host_definer_manager.py | 1 - .../kubernetes_manager/manager.py | 13 ++ controllers/servers/host_definer/messages.py | 6 +- .../storage_manager/host_definer_server.py | 29 ++--- .../host_definer/watcher/csi_node_watcher.py | 15 ++- .../watcher/host_definition_watcher.py | 7 ++ .../watcher/storage_class_watcher.py | 1 + .../host_definer/watcher/watcher_helper.py | 29 ++++- controllers/servers/messages.py | 2 +- controllers/servers/settings.py | 1 - controllers/servers/utils.py | 58 ++++++++- node/pkg/driver/node.go | 3 +- node/pkg/driver/node_test.go | 26 ++-- node/pkg/driver/node_utils.go | 115 +++++++++++------- node/pkg/driver/node_utils_test.go | 96 --------------- 22 files changed, 233 insertions(+), 229 deletions(-) diff --git a/common/config.yaml b/common/config.yaml index 7e235dc66..ed010ab3e 100644 --- a/common/config.yaml +++ b/common/config.yaml @@ -20,9 +20,6 @@ parameters: object_id_info: delimiter: ":" ids_delimiter: ";" - node_id_info: - delimiter: ";" - fcs_delimiter: ":" connectivity_type: nvme_over_fc: "nvmeofc" diff --git a/controllers/common/node_info.py b/controllers/common/node_info.py index 08da81b33..a710c081b 100644 --- a/controllers/common/node_info.py +++ b/controllers/common/node_info.py @@ -1,26 +1,19 @@ from dataclasses import dataclass, field -import controllers.servers.settings as servers_settings from controllers.array_action import settings as array_config -from controllers.common import utils class NodeIdInfo: def __init__(self, node_id): - """ - Args: - node_id: ,, - """ - node_name, nvme_nqn, fc_wwns_str, iscsi_iqn = utils.get_node_id_info(node_id) - fc_wwns = fc_wwns_str.split(servers_settings.PARAMETERS_FC_WWN_DELIMITER) - self.node_name = node_name - self.initiators = Initiators([nvme_nqn], fc_wwns, [iscsi_iqn]) + self.node_name = node_id @dataclass class Initiators: """ Object containing node initiators (e.g. iqn, fc_wwns) + Usage: + inits = Initiators([nvme_nqn], fc_wwns, [iscsi_iqn]) """ nvme_nqns: list = field(default_factory=list) fc_wwns: list = field(default_factory=list) diff --git a/controllers/common/settings.py b/controllers/common/settings.py index 686dbfbae..4e7d8c7ef 100644 --- a/controllers/common/settings.py +++ b/controllers/common/settings.py @@ -1,5 +1,3 @@ -from controllers.common.config import config - NAME_PREFIX_SEPARATOR = "_" ENDPOINTS_SEPARATOR = ", " @@ -11,7 +9,6 @@ ARRAY_TYPE_DS8K = 'DS8K' ALL_ARRAY_TYPES = [ARRAY_TYPE_XIV, ARRAY_TYPE_SVC, ARRAY_TYPE_DS8K] -PARAMETERS_NODE_ID_DELIMITER = config.parameters.node_id_info.delimiter SPACE_EFFICIENCY_THIN = "thin" SPACE_EFFICIENCY_COMPRESSED = "compressed" SPACE_EFFICIENCY_DEDUPLICATED = "deduplicated" @@ -26,9 +23,12 @@ HOST_DEFINITION_NODE_ID_FIELD = 'nodeId' NAME_FIELD = 'name' NAMESPACE_FIELD = 'namespace' +ANNOTATIONS_FIELD = 'annotations' IO_GROUP_DELIMITER = ':' IO_GROUP_LABEL_PREFIX = 'hostdefiner.block.csi.ibm.com/io-group-' +NODE_INITIATORS_FIELD = 'block.csi.ibm.com/node-initiators' + EAR_VOLUME_FC_MAP_COUNT = "2" SCSI_PROTOCOL = 'scsi' diff --git a/controllers/common/utils.py b/controllers/common/utils.py index 5b52c879a..838921ce2 100644 --- a/controllers/common/utils.py +++ b/controllers/common/utils.py @@ -11,9 +11,7 @@ import urllib3 -from controllers.common import settings from controllers.common.csi_logger import get_stdout_logger -from controllers.servers import messages logger = get_stdout_logger() @@ -44,23 +42,6 @@ def string_to_array(str_val, separator): return res -def get_node_id_info(node_id): - logger.debug("getting node info for node id : {0}".format(node_id)) - split_node = node_id.split(settings.PARAMETERS_NODE_ID_DELIMITER) - hostname, nvme_nqn, fc_wwns, iscsi_iqn = "", "", "", "" - if len(split_node) == 4: - hostname, nvme_nqn, fc_wwns, iscsi_iqn = split_node - elif len(split_node) == 3: - hostname, nvme_nqn, fc_wwns = split_node - elif len(split_node) == 2: - hostname, nvme_nqn = split_node - else: - raise ValueError(messages.WRONG_FORMAT_MESSAGE.format("node id")) - logger.debug("node name : {0}, nvme_nqn: {1}, fc_wwns : {2}, iscsi_iqn : {3} ".format( - hostname, nvme_nqn, fc_wwns, iscsi_iqn)) - return hostname, nvme_nqn, fc_wwns, iscsi_iqn - - def _get_config_map_info(): result = {} diff --git a/controllers/servers/csi/csi_controller_server.py b/controllers/servers/csi/csi_controller_server.py index 51638091d..00bfa22f3 100755 --- a/controllers/servers/csi/csi_controller_server.py +++ b/controllers/servers/csi/csi_controller_server.py @@ -229,7 +229,7 @@ def ControllerPublishVolume(self, request, context): volume_id = volume_id_info.ids.uid node_id_info = NodeIdInfo(request.node_id) node_name = node_id_info.node_name - initiators = node_id_info.initiators + initiators = utils.get_node_initiators(node_name) logger.debug("node name for this publish operation is : {0}".format(node_name)) @@ -273,7 +273,7 @@ def ControllerUnpublishVolume(self, request, context): volume_id = volume_id_info.ids.uid node_id_info = NodeIdInfo(request.node_id) node_name = node_id_info.node_name - initiators = node_id_info.initiators + initiators = utils.get_node_initiators(node_name) logger.debug("node name for this unpublish operation is : {0}".format(node_name)) array_connection_info = utils.get_array_connection_info_from_secrets(request.secrets, diff --git a/controllers/servers/csi/requirements.txt b/controllers/servers/csi/requirements.txt index b76974d15..522aea8d1 100644 --- a/controllers/servers/csi/requirements.txt +++ b/controllers/servers/csi/requirements.txt @@ -6,6 +6,7 @@ munch==4.0.0 retry2==0.9.5 packaging==25.0 base58==2.1.1 +kubernetes==33.1.0 # A9000 python client pyxcli==1.2.1 diff --git a/controllers/servers/host_definer/hd_types.py b/controllers/servers/host_definer/hd_types.py index 4d8ec9b6e..106548b1c 100644 --- a/controllers/servers/host_definer/hd_types.py +++ b/controllers/servers/host_definer/hd_types.py @@ -8,7 +8,9 @@ class DefineHostRequest: prefix: str = '' connectivity_type_from_user: str = '' node_id_from_host_definition: str = '' + node_initiators_from_host_definition: str = '' node_id_from_csi_node: str = '' + node_initiators_from_csi_node: str = '' array_connection_info: ArrayConnectionInfo = ArrayConnectionInfo(array_addresses='', user='', password='', partition_name='', partition_vg='', port_set='') io_group: str = '' @@ -18,7 +20,8 @@ class DefineHostRequest: class DefineHostResponse: error_message: str = '' connectivity_type: str = '' - ports: list = field(default_factory=list) + ports: list = field(default_factory=list) # The ports + # that actually found in storage (see _generate_response in host_definer_server.py) node_name_on_storage: str = '' io_group: list = field(default_factory=list) management_address: str = '' @@ -40,6 +43,7 @@ class HostDefinitionInfo: secret_namespace: str = '' node_name: str = '' node_id: str = '' + node_initiators: str = '' phase: str = '' resource_version: str = '' uid: str = '' @@ -50,8 +54,15 @@ class HostDefinitionInfo: @dataclass class CsiNodeInfo: + """ + class to store CsiNodeInfo. + Typically - name and node_id can be extracted from csinode manifest + node_initiators should be extracted from the node annotations. + (See function: _generate_csi_node_info) + """ name: str = '' node_id: str = '' + node_initiators: str = '' @dataclass @@ -77,4 +88,5 @@ class ManagedNode: def __init__(self, csi_node_info, labels): self.name = csi_node_info.name self.node_id = csi_node_info.node_id + self.node_initiators = csi_node_info.node_initiators self.io_group = utils.generate_io_group_from_labels(labels) diff --git a/controllers/servers/host_definer/host_definer_manager.py b/controllers/servers/host_definer/host_definer_manager.py index 1f4ee0459..1ba98b25f 100644 --- a/controllers/servers/host_definer/host_definer_manager.py +++ b/controllers/servers/host_definer/host_definer_manager.py @@ -21,7 +21,6 @@ def __init__(self): self.node_watcher = NodeWatcher() def start_host_definition(self): - logger.info('starting host definer') self.csi_node_watcher.add_initial_csi_nodes() self.storage_class_watcher.add_initial_storage_classes() self.node_watcher.add_initial_nodes() diff --git a/controllers/servers/host_definer/kubernetes_manager/manager.py b/controllers/servers/host_definer/kubernetes_manager/manager.py index 9ec515f90..f13687949 100644 --- a/controllers/servers/host_definer/kubernetes_manager/manager.py +++ b/controllers/servers/host_definer/kubernetes_manager/manager.py @@ -1,3 +1,4 @@ +"docstring" import ast import datetime import base64 @@ -12,11 +13,13 @@ import controllers.common.settings as common_settings from controllers.servers.host_definer.hd_types import ( CsiNodeInfo, PodInfo, NodeInfo, StorageClassInfo, HostDefinitionInfo) +from controllers.servers import utils as controllers_utils logger = get_stdout_logger() class KubernetesManager(): + """docstring""" def __init__(self): self._load_cluster_configuration() self.dynamic_client = self._get_dynamic_client() @@ -109,6 +112,7 @@ def _generate_csi_node_info(self, k8s_csi_node): csi_node_info = CsiNodeInfo() csi_node_info.name = k8s_csi_node.metadata.name csi_node_info.node_id = self._get_node_id_from_k8s_csi_node(k8s_csi_node) + csi_node_info.node_initiators = controllers_utils.get_node_initiators_data(csi_node_info.name) return csi_node_info def _get_node_id_from_k8s_csi_node(self, k8s_csi_node): @@ -148,8 +152,12 @@ def _generate_host_definition_info(self, k8s_host_definition): k8s_host_definition, settings.NODE_NAME_FIELD) host_definition_info.node_id = self._get_attr_from_host_definition( k8s_host_definition, common_settings.HOST_DEFINITION_NODE_ID_FIELD) + host_definition_info.node_initiators = self._get_attr_from_host_definition_annotations( + k8s_host_definition, common_settings.NODE_INITIATORS_FIELD) host_definition_info.connectivity_type = self._get_attr_from_host_definition( k8s_host_definition, settings.CONNECTIVITY_TYPE_FIELD) + host_definition_info.ports = self._get_attr_from_host_definition( + k8s_host_definition, settings.PORTS_FIELD) return host_definition_info def _get_k8s_object_resource_version(self, k8s_object): @@ -167,6 +175,11 @@ def _get_attr_from_host_definition(self, k8s_host_definition, attribute): return getattr(k8s_host_definition.spec.hostDefinition, attribute) return '' + def _get_attr_from_host_definition_annotations(self, k8s_host_definition, attribute): + if hasattr(k8s_host_definition.metadata.annotations, attribute): + return getattr(k8s_host_definition.metadata.annotations, attribute) + return '' + def _is_host_definition_matches(self, host_definition_info, node_name, secret_name, secret_namespace): return host_definition_info.node_name == node_name and \ host_definition_info.secret_name == secret_name and \ diff --git a/controllers/servers/host_definer/messages.py b/controllers/servers/host_definer/messages.py index e0a60f24a..618606bee 100644 --- a/controllers/servers/host_definer/messages.py +++ b/controllers/servers/host_definer/messages.py @@ -38,8 +38,8 @@ ADD_FINALIZER_TO_HOST_DEFINITION = 'Adding finalizer to host definition {}' REMOVE_FINALIZER_TO_HOST_DEFINITION = 'Removing finalizer from host definition {}' FAILED_TO_REMOVE_FINALIZER = 'Failed to remove {} finalizer from node'.format(settings.CSI_IBM_FINALIZER) -NODE_ID_WAS_CHANGED = "NodeId was changed for {} node, updating his ports in his definitions,"\ - "old NodeId [{}], new NodeId [{}]" +NODE_INITIATORS_WERE_CHANGED = "Node initiators were changed for {} node, updating its ports definitions, " \ + "old initiators: {}, new initiators: {}" UPDATE_HOST_DEFINITION_FIELDS_FROM_STORAGE = 'Update host definition {} host from storage fields with {}' CHECK_NODE_SHOULD_BE_MANAGED_BY_SECRET = 'Check if node {} should be managed by {} secret in {} namespace' NODE_SHOULD_BE_MANAGED_ON_SECRET = 'Node {} should be managed by {} secret in {} namespace' @@ -75,3 +75,5 @@ GENERATE_REQUEST_FOR_NODE = 'Generating define request for node {}' COULD_NOT_CHANGE_HOST_PROTOCOL_USING_CHHOST = "Could not change host [{}] protocol using chhost command, "\ "changing the host protocol by deleting it and creating it again" +NODE_IS_NOT_MANAGED_NODE_TYPE = "Node {} is not a ManagedNode instance. This might lead to attribute errors." +NODE_ID_IS_NONE = "node_id is None !!!" diff --git a/controllers/servers/host_definer/storage_manager/host_definer_server.py b/controllers/servers/host_definer/storage_manager/host_definer_server.py index efb1cc409..2d0df5606 100644 --- a/controllers/servers/host_definer/storage_manager/host_definer_server.py +++ b/controllers/servers/host_definer/storage_manager/host_definer_server.py @@ -1,3 +1,6 @@ +"""host_definer_server.py""" +import json + from controllers.array_action import settings as array_config from controllers.array_action.errors import HostNotFoundError, HostAlreadyExists from controllers.array_action.storage_agent import detect_array_type, get_agent @@ -5,8 +8,9 @@ from controllers.common.node_info import NodeIdInfo import controllers.common.settings as common_settings from controllers.servers.host_definer.hd_types import DefineHostResponse -from controllers.servers.utils import join_object_prefix_with_name, get_initiators_connectivity_type -import controllers.servers.host_definer.settings as host_definer_settings +from controllers.servers.utils import (join_object_prefix_with_name, + get_initiators_connectivity_type, + generate_node_initiators_from_string_data) from controllers.servers.host_definer import messages logger = get_stdout_logger() @@ -17,8 +21,8 @@ def define_host(self, request): array_connection_info = request.array_connection_info array_addresses = array_connection_info.array_addresses node_id_info = NodeIdInfo(request.node_id_from_csi_node) - initiators = node_id_info.initiators node_name = node_id_info.node_name + initiators = generate_node_initiators_from_string_data(request.node_initiators_from_csi_node) connectivity_type_from_user = get_initiators_connectivity_type(initiators, request.connectivity_type_from_user) host_name = join_object_prefix_with_name(prefix=request.prefix, name=node_name) logger.info(messages.DEFINE_NODE_ON_ARRAYS.format(node_name, array_addresses)) @@ -26,8 +30,8 @@ def define_host(self, request): array_type = detect_array_type(array_addresses) with get_agent(array_connection_info, array_type).get_mediator() as array_mediator: try: - initiators_from_host_definition = self._get_initiators_from_node_id( - request.node_id_from_host_definition) + initiators_from_host_definition = \ + generate_node_initiators_from_string_data(request.node_initiators_from_host_definition) found_host_name = self._get_host_name(initiators_from_host_definition, array_mediator) # Partition update is first one - verifies partition can be fixed (may fail if mapped) self._update_host_partition(request, found_host_name, @@ -56,9 +60,9 @@ def define_host(self, request): def undefine_host(self, request): node_id_info = NodeIdInfo(request.node_id_from_csi_node) + initiators = generate_node_initiators_from_string_data(request.node_initiators_from_csi_node) array_connection_info = request.array_connection_info array_addresses = array_connection_info.array_addresses - initiators = node_id_info.initiators node_name = node_id_info.node_name logger.info(messages.UNDEFINE_NODE_FROM_ARRAYS.format(node_name, array_addresses)) try: @@ -88,7 +92,7 @@ def _update_host_partition(self, request, host_name, partition_name, array_media raise array_errors.HostNotFoundError(host_name) def _update_host_ports(self, request, host, array_mediator, partition_name): - initiators = self._get_initiators_from_node_id(request.node_id_from_csi_node) + initiators = generate_node_initiators_from_string_data(request.node_initiators_from_csi_node) connectivity_type_from_user = get_initiators_connectivity_type(initiators, request.connectivity_type_from_user) connectivity_type_from_host = array_mediator.get_host_connectivity_type(host) if self._is_protocol_switched(connectivity_type_from_user, connectivity_type_from_host): @@ -108,10 +112,6 @@ def _update_host_ports(self, request, host, array_mediator, partition_name): return raise ex - def _get_initiators_from_node_id(self, node_id): - node_id_info = NodeIdInfo(node_id) - return node_id_info.initiators - def _is_protocol_switched(self, connectivity_type_from_user, connectivity_type_from_host): return self._is_switching_from_nvme_to_scsi(connectivity_type_from_user, connectivity_type_from_host) or \ self._is_switching_from_scsi_to_nvme(connectivity_type_from_user, connectivity_type_from_host) @@ -136,7 +136,7 @@ def _change_host_protocol(self, array_mediator, host_name, connectivity_type_fro def _change_host_protocol_with_chhost(self, array_mediator, host_name, connectivity_type_from_host, request): self._remove_host_ports(array_mediator, host_name, connectivity_type_from_host) - initiators = self._get_initiators_from_node_id(request.node_id_from_csi_node) + initiators = generate_node_initiators_from_string_data(request.node_initiators_from_csi_node) connectivity_type_from_user = get_initiators_connectivity_type(initiators, request.connectivity_type_from_user) protocol = self._get_host_protocol(connectivity_type_from_user) array_mediator.change_host_protocol(host_name, protocol) @@ -161,7 +161,7 @@ def _is_protocol_nvme(self, connectivity_type): return connectivity_type == array_config.NVME_OVER_FC_CONNECTIVITY_TYPE def _create_host(self, host, array_mediator, request): - initiators = self._get_initiators_from_node_id(request.node_id_from_csi_node) + initiators = generate_node_initiators_from_string_data(request.node_initiators_from_csi_node) connectivity_type = get_initiators_connectivity_type(initiators, request.connectivity_type_from_user) array_mediator.create_host(host, initiators, connectivity_type, request.io_group, request.array_connection_info.partition_name, @@ -173,7 +173,8 @@ def _create_host(self, host, array_mediator, request): def _is_port_update_needed_when_same_protocol( self, request, connectivity_type_from_user, connectivity_type_from_host): return connectivity_type_from_user != connectivity_type_from_host \ - or request.node_id_from_csi_node != request.node_id_from_host_definition + or request.node_id_from_csi_node != request.node_id_from_host_definition \ + or request.node_initiators_from_csi_node != request.node_initiators_from_host_definition def _update_host_io_group(self, request, host, array_mediator): io_group_from_host = array_mediator.get_host_io_group(host) diff --git a/controllers/servers/host_definer/watcher/csi_node_watcher.py b/controllers/servers/host_definer/watcher/csi_node_watcher.py index 7c47eb58c..459bed02c 100644 --- a/controllers/servers/host_definer/watcher/csi_node_watcher.py +++ b/controllers/servers/host_definer/watcher/csi_node_watcher.py @@ -93,15 +93,18 @@ def _create_definitions_when_csi_node_changed(self, csi_node_info): host_definition_info = self._get_matching_host_definition_info( csi_node_info.name, secret_name, secret_namespace) if host_definition_info: - if self._is_node_id_changed(host_definition_info.node_id, csi_node_info.node_id): - logger.info(messages.NODE_ID_WAS_CHANGED.format(csi_node_info.name, - host_definition_info.node_id, csi_node_info.node_id)) + if self._is_node_initiators_changed(host_definition_info, csi_node_info): + logger.info(messages.NODE_INITIATORS_WERE_CHANGED.format(csi_node_info.name, + host_definition_info.node_initiators, csi_node_info.node_initiators)) NODES[csi_node_info.name] = self._generate_managed_node(csi_node_info) self._create_definition(host_definition_info) - def _is_node_id_changed(self, host_definition_node_id, csi_node_node_id): - return host_definition_node_id != csi_node_node_id \ - and host_definition_node_id and csi_node_node_id + def _is_node_initiators_changed(self, host_definition_info, csi_node_info): + if not host_definition_info.node_id or not csi_node_info.node_id: + logger.warning(messages.NODE_ID_IS_NONE) + return False + + return host_definition_info.node_initiators != csi_node_info.node_initiators def _undefine_hosts(self, node_name): for secret_info in MANAGED_SECRETS: diff --git a/controllers/servers/host_definer/watcher/host_definition_watcher.py b/controllers/servers/host_definer/watcher/host_definition_watcher.py index cb33f6065..a35b90122 100644 --- a/controllers/servers/host_definer/watcher/host_definition_watcher.py +++ b/controllers/servers/host_definer/watcher/host_definition_watcher.py @@ -78,6 +78,7 @@ def _get_action(self, phase): return settings.UNDEFINE_ACTION def _define_host_after_pending(self, host_definition_info): + logger.info(host_definition_info) response = DefineHostResponse() if self._is_node_should_be_managed_on_secret( host_definition_info.node_name, host_definition_info.secret_name, @@ -86,6 +87,7 @@ def _define_host_after_pending(self, host_definition_info): self._update_host_definition_from_storage_response(host_definition_info.name, response) else: self._delete_host_definition(host_definition_info.name) + logger.info(response) return response def _update_host_definition_from_storage_response(self, host_definition_name, response): @@ -94,6 +96,11 @@ def _update_host_definition_from_storage_response(self, host_definition_name, re self._patch_host_definition(host_definition_manifest) def _generate_host_definition_manifest(self, host_definition_name, response): + """ + This method constructs the subset of the manifest required for a PATCH request, + To generate a full HostDefinition manifest, + use function: _get_host_definition_manifest in `watcher_helper.py`. + """ return { settings.METADATA: { common_settings.NAME_FIELD: host_definition_name, diff --git a/controllers/servers/host_definer/watcher/storage_class_watcher.py b/controllers/servers/host_definer/watcher/storage_class_watcher.py index 9ea9a08be..46a9c7a05 100644 --- a/controllers/servers/host_definer/watcher/storage_class_watcher.py +++ b/controllers/servers/host_definer/watcher/storage_class_watcher.py @@ -14,6 +14,7 @@ class StorageClassWatcher(Watcher): def add_initial_storage_classes(self): storage_classes_info = self._get_storage_classes_info() + logger.info(storage_classes_info) for storage_class_info in storage_classes_info: secrets_info = self._get_secrets_info_from_storage_class_with_driver_provisioner(storage_class_info) self._handle_added_watch_event(secrets_info, storage_class_info.name) diff --git a/controllers/servers/host_definer/watcher/watcher_helper.py b/controllers/servers/host_definer/watcher/watcher_helper.py index b972f7b82..b3184f692 100644 --- a/controllers/servers/host_definer/watcher/watcher_helper.py +++ b/controllers/servers/host_definer/watcher/watcher_helper.py @@ -38,6 +38,7 @@ def _define_host_on_all_storages(self, node_name): def _get_host_definition_info_from_secret_and_node_name(self, node_name, secret_info): host_definition_info = self._get_host_definition_info_from_secret(secret_info) host_definition_info = self._add_name_to_host_definition_info(node_name, host_definition_info) + host_definition_info = self._add_node_initiators_to_host_definition_info(node_name, host_definition_info) return host_definition_info def _get_host_definition_info_from_secret(self, secret_info): @@ -49,6 +50,7 @@ def _get_host_definition_info_from_secret(self, secret_info): def _define_nodes(self, host_definition_info): for node_name, _ in NODES.items(): host_definition_info = self._add_name_to_host_definition_info(node_name, host_definition_info) + host_definition_info = self._add_node_initiators_to_host_definition_info(node_name, host_definition_info) self._create_definition(host_definition_info) def _add_name_to_host_definition_info(self, node_name, host_definition_info): @@ -57,6 +59,13 @@ def _add_name_to_host_definition_info(self, node_name, host_definition_info): host_definition_info.name = self._get_host_definition_name(node_name) return host_definition_info + def _add_node_initiators_to_host_definition_info(self, node_name, host_definition_info): + node_obj = NODES[node_name] + if not isinstance(node_obj, ManagedNode): + logger.warning(messages.NODE_IS_NOT_MANAGED_NODE_TYPE.format(node_name)) + host_definition_info.node_initiators = node_obj.node_initiators + return host_definition_info + def _create_definition(self, host_definition_info): if not self._is_node_should_be_managed_on_secret( host_definition_info.node_name, host_definition_info.secret_name, @@ -75,6 +84,7 @@ def _update_host_definition_info(self, host_definition_info): if host_definition_info_on_cluster: host_definition_info.connectivity_type = host_definition_info_on_cluster.connectivity_type host_definition_info.node_id = host_definition_info_on_cluster.node_id + host_definition_info.node_initiators = host_definition_info_on_cluster.node_initiators return host_definition_info def _define_host(self, host_definition_info): @@ -86,11 +96,13 @@ def _create_host_definition_if_not_exist(self, host_definition_info, response): host_definition_manifest = self._get_host_definition_manifest(host_definition_info, response) current_host_definition_info_on_cluster = self._get_matching_host_definition_info( host_definition_info.node_name, host_definition_info.secret_name, host_definition_info.secret_namespace) + logger.info(host_definition_manifest) + logger.info(current_host_definition_info_on_cluster) if current_host_definition_info_on_cluster: host_definition_manifest[settings.METADATA][ common_settings.NAME_FIELD] = current_host_definition_info_on_cluster.name self._patch_host_definition(host_definition_manifest) - return current_host_definition_info_on_cluster + return current_host_definition_info_on_cluster # See CSI-6058 else: logger.info(messages.CREATING_NEW_HOST_DEFINITION.format(host_definition_info.name)) return self._create_host_definition(host_definition_manifest) @@ -101,6 +113,9 @@ def _get_host_definition_manifest(self, host_definition_info, response): settings.KIND: settings.HOST_DEFINITION_KIND, settings.METADATA: { common_settings.NAME_FIELD: host_definition_info.name, + common_settings.ANNOTATIONS_FIELD: { + common_settings.NODE_INITIATORS_FIELD: NODES[host_definition_info.node_name].node_initiators + } }, settings.SPEC: { settings.HOST_DEFINITION_FIELD: { @@ -238,6 +253,7 @@ def _ensure_definition_state(self, host_definition_info, define_function): return define_function(request) def _get_request_from_host_definition(self, host_definition_info): + logger.info(host_definition_info) node_name = host_definition_info.node_name logger.info(messages.GENERATE_REQUEST_FOR_NODE.format(node_name)) node_info = self._get_node_info(node_name) @@ -246,7 +262,12 @@ def _get_request_from_host_definition(self, host_definition_info): request, host_definition_info.secret_name, host_definition_info.secret_namespace, node_info.labels) if request: request.node_id_from_host_definition = host_definition_info.node_id + request.node_initiators_from_host_definition = host_definition_info.node_initiators request.node_id_from_csi_node = self._get_node_id_by_node(host_definition_info) + request.node_initiators_from_csi_node = self._get_node_initiators_by_node(host_definition_info) + # To clarify the variable name 'node_initiators_from_csi_node' - + # Since CSI-5997, initiators are extracted from k8s node annoatations (not from k&s csinode nodeID). + # However, They are still stored at CsiNodeInfo and ManagedNode (see hd_types.py) request.io_group = self._get_io_group_by_node(host_definition_info.node_name) return request @@ -321,6 +342,12 @@ def _get_node_id_by_node(self, host_definition_info): except Exception: return host_definition_info.node_id + def _get_node_initiators_by_node(self, host_definition_info): + try: + return NODES[host_definition_info.node_name].node_initiators + except Exception: + return host_definition_info.node_initiators + def _get_io_group_by_node(self, node_name): try: return NODES[node_name].io_group diff --git a/controllers/servers/messages.py b/controllers/servers/messages.py index e54d39b98..2354bd7ec 100644 --- a/controllers/servers/messages.py +++ b/controllers/servers/messages.py @@ -1,6 +1,6 @@ VALIDATION_EXCEPTION_MESSAGE = "Validation error has occurred : {0}" -WRONG_ID_FORMAT_MESSAGE = "Wrong {0} id format : {1}" +WRONG_ID_FORMAT_MESSAGE = "Wrong {0} id format or type : {1}" OBJECT_ALREADY_PROCESSING_MESSAGE = "object {0} is already processing. request cannot be completed." diff --git a/controllers/servers/settings.py b/controllers/servers/settings.py index 17376251e..92525761f 100644 --- a/controllers/servers/settings.py +++ b/controllers/servers/settings.py @@ -52,7 +52,6 @@ PARAMETERS_CAPABILITIES_DELIMITER = "=" PARAMETERS_OBJECT_ID_INFO_DELIMITER = config.parameters.object_id_info.delimiter PARAMETERS_OBJECT_IDS_DELIMITER = config.parameters.object_id_info.ids_delimiter -PARAMETERS_FC_WWN_DELIMITER = config.parameters.node_id_info.fcs_delimiter PARAMETERS_TOPOLOGY_DELIMITER = "/" PARAMETERS_ARRAY_ADDRESSES_DELIMITER = "," diff --git a/controllers/servers/utils.py b/controllers/servers/utils.py index 2c8fb85cb..d8c1a2493 100644 --- a/controllers/servers/utils.py +++ b/controllers/servers/utils.py @@ -24,6 +24,8 @@ ObjectIdInfo, ObjectParameters, VolumeGroupParameters, VolumeGroupIdInfo) from controllers.servers.errors import ObjectIdError, ValidationException, InvalidNodeId +from controllers.common.node_info import Initiators +from controllers.servers.host_definer.kubernetes_manager.manager import KubernetesManager logger = get_stdout_logger() @@ -92,6 +94,48 @@ def _get_array_connection_info_from_system_info(secrets, system_id): partition_name=partition_name, partition_vg=partition_vg, port_set=port_set) +def get_node_initiators_data(node_name): + """ + Return value example: + '{"fc":[],"iscsi":["iqn.2016-04.com.open-iscsi:8bce7b6eab12"],"nvme":[]}' + """ + initiators_data = '' + kubernetes_manager = KubernetesManager() + core_api = kubernetes_manager.core_api + k8s_node = core_api.read_node(name=node_name) + if k8s_node: + node_annotations = k8s_node.metadata.annotations + initiators_data = node_annotations.get(settings.NODE_INITIATORS_FIELD, "{}") + return initiators_data + + +def generate_node_initiators_from_string_data(initiators_data): + """ + Return value example: + Initiators(nvme_nqns=[], fc_wwns=[], iscsi_iqns=['iqn.2016-04.com.open-iscsi:8bce7b6eab12']) + """ + initiators_data = json.loads(initiators_data) + nvme_nqns = initiators_data.get("nvme", []) + fc_wwns = initiators_data.get("fc", []) + iscsi_iqns = initiators_data.get("iscsi", []) + return Initiators(nvme_nqns, fc_wwns, iscsi_iqns) + + +def get_node_initiators_from_k8s_node(k8s_node): + "docstring" + node_annotations = k8s_node.metadata.annotations + initiators_data = node_annotations.get(settings.NODE_INITIATORS_FIELD, "{}") + initiators = generate_node_initiators_from_string_data(initiators_data) + return initiators + + +def get_node_initiators(node_name): + "docstring" + initiators_data = get_node_initiators_data(node_name) + initiators = generate_node_initiators_from_string_data(initiators_data) + return initiators + + def get_array_connection_info_from_secrets(secrets, topologies=None, system_id=None): system_info, system_id = _get_system_info_from_secrets(secrets, topologies, system_id) return _get_array_connection_info_from_system_info(system_info, system_id) @@ -514,11 +558,15 @@ def validate_delete_volume_request(request): def _validate_node_id(node_id): - logger.debug("validating node id") - - delimiter_count = node_id.count(settings.PARAMETERS_NODE_ID_DELIMITER) - - if not 1 <= delimiter_count <= 3: + """ + In the past, this function validated that node_id format is: + 'node_name;nvme0;fc0:fc1:fc2:fc3;iscsi0' + Since CSI-5997, node_id contains only node_name. + Hence, we only validate type and len. + """ + logger.debug("validating node id: %s", node_id) + + if not isinstance(node_id, str) or len(node_id) < 1: raise InvalidNodeId(node_id) logger.debug("node id validation finished") diff --git a/node/pkg/driver/node.go b/node/pkg/driver/node.go index d9f6bcc76..bd3239486 100644 --- a/node/pkg/driver/node.go +++ b/node/pkg/driver/node.go @@ -867,7 +867,8 @@ func (d *NodeService) NodeGetInfo(ctx context.Context, req *csi.NodeGetInfoReque return nil, status.Error(codes.Internal, err.Error()) } - nodeId, err := d.NodeUtils.GenerateNodeID(d.Hostname, nvmeNQN, fcWWNs, iscsiIQN) + var nodeId = d.Hostname + err = d.NodeUtils.UpdateNodePortsAnnotation(ctx, d.Hostname, iscsiIQN, fcWWNs, nvmeNQN) if err != nil { return nil, status.Error(codes.Internal, err.Error()) } diff --git a/node/pkg/driver/node_test.go b/node/pkg/driver/node_test.go index 8e4204d08..7867a30aa 100644 --- a/node/pkg/driver/node_test.go +++ b/node/pkg/driver/node_test.go @@ -1393,7 +1393,7 @@ func TestNodeGetInfo(t *testing.T) { name: "empty nqn with error from node_utils, valid fcs", returnNqnErr: fmt.Errorf("some error"), returnFcs: []string{"10000000c9934d9f", "10000000c9934d9h"}, - expNodeId: "test-host;;10000000c9934d9f:10000000c9934d9h", + expNodeId: "test-host", nvmeExists: true, fcExists: true, iscsiExists: true, @@ -1410,7 +1410,7 @@ func TestNodeGetInfo(t *testing.T) { name: "empty iqn with error from node_utils, valid fcs", returnIqnErr: fmt.Errorf("some error"), returnFcs: []string{"10000000c9934d9f", "10000000c9934d9h"}, - expNodeId: "test-host;;10000000c9934d9f:10000000c9934d9h", + expNodeId: "test-host", nvmeExists: false, fcExists: true, iscsiExists: true, @@ -1419,7 +1419,7 @@ func TestNodeGetInfo(t *testing.T) { name: "valid iqn and fcs", returnIqn: "iqn.1994-07.com.redhat:e123456789", returnFcs: []string{"10000000c9934d9f", "10000000c9934d9h"}, - expNodeId: "test-host;;10000000c9934d9f:10000000c9934d9h;iqn.1994-07.com.redhat:e123456789", + expNodeId: "test-host", nvmeExists: false, fcExists: true, iscsiExists: true, @@ -1429,7 +1429,7 @@ func TestNodeGetInfo(t *testing.T) { returnNqn: "nqn.2014-08.org.nvmexpress:uuid:b57708c7-5bb6-46a0-b2af-9d824bf539e1", returnFcs: []string{"10000000c9934d9f", "10000000c9934d9h"}, returnIqn: "iqn.1994-07.com.redhat:e123456789", - expNodeId: "test-host;nqn.2014-08.org.nvmexpress:uuid:b57708c7-5bb6-46a0-b2af-9d824bf539e1;10000000c9934d9f:10000000c9934d9h;iqn.1994-07.com.redhat:e123456789", + expNodeId: "test-host", nvmeExists: true, fcExists: true, iscsiExists: true, @@ -1447,7 +1447,7 @@ func TestNodeGetInfo(t *testing.T) { fcExists: true, iscsiExists: false, returnFcs: []string{"10000000c9934d9f"}, - expNodeId: "test-host;;10000000c9934d9f", + expNodeId: "test-host", }, { name: "fc path is inexistent", @@ -1456,16 +1456,7 @@ func TestNodeGetInfo(t *testing.T) { iscsiExists: true, returnNqn: "nqn.2014-08.org.nvmexpress:uuid:b57708c7-5bb6-46a0-b2af-9d824bf539e1", returnIqn: "iqn.1994-07.com.redhat:e123456789", - expNodeId: "test-host;nqn.2014-08.org.nvmexpress:uuid:b57708c7-5bb6-46a0-b2af-9d824bf539e1;;iqn.1994-07.com.redhat:e123456789", - }, { - name: "generate NodeID returns error", - returnIqn: "iqn.1994-07.com.redhat:e123456789", - returnFcs: []string{"10000000c9934d9f", "10000000c9934d9h"}, - returnNodeIdErr: fmt.Errorf("some error"), - expErr: status.Error(codes.Internal, fmt.Errorf("some error").Error()), - nvmeExists: false, - fcExists: true, - iscsiExists: true, + expNodeId: "test-host", }, } for _, tc := range testCases { @@ -1477,6 +1468,7 @@ func TestNodeGetInfo(t *testing.T) { fakeNodeutils := mocks.NewMockNodeUtilsInterface(mockCtrl) d := newTestNodeService(fakeNodeutils, nil, nil) + fakeNodeutils.EXPECT().UpdateNodePortsAnnotation(context.TODO(), d.Hostname, tc.returnIqn, tc.returnFcs, tc.returnNqn).AnyTimes() fakeNodeutils.EXPECT().GetTopologyLabels(context.TODO(), d.Hostname).Return(topologySegments, nil) fakeNodeutils.EXPECT().IsPathExists(driver.NvmeFullPath).Return(tc.nvmeExists) fakeNodeutils.EXPECT().IsFCExists().Return(tc.fcExists) @@ -1493,10 +1485,6 @@ func TestNodeGetInfo(t *testing.T) { } } - if tc.returnNqn != "" || len(tc.returnFcs) > 0 || tc.returnIqn != "" { - fakeNodeutils.EXPECT().GenerateNodeID("test-host", tc.returnNqn, tc.returnFcs, tc.returnIqn).Return(tc.expNodeId, tc.returnNodeIdErr) - } - expTopology := &csi.Topology{Segments: topologySegments} expResponse := &csi.NodeGetInfoResponse{NodeId: tc.expNodeId, AccessibleTopology: expTopology} diff --git a/node/pkg/driver/node_utils.go b/node/pkg/driver/node_utils.go index 0698b6dcf..5de7f6b99 100644 --- a/node/pkg/driver/node_utils.go +++ b/node/pkg/driver/node_utils.go @@ -19,6 +19,7 @@ package driver import ( "bufio" "context" + "encoding/json" "fmt" "io" "io/ioutil" @@ -29,6 +30,7 @@ import ( "github.com/ibm/ibm-block-csi-driver/node/pkg/driver/device_connectivity" "golang.org/x/sys/unix" + "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -42,6 +44,7 @@ import ( var ( getOpts = metav1.GetOptions{} + patchOpts = metav1.PatchOptions{} topologyPrefixes = [...]string{"topology.block.csi.ibm.com"} ) @@ -88,8 +91,8 @@ type NodeUtilsInterface interface { FormatDevice(devicePath string, fsType string) IsNotMountPoint(file string) (bool, error) GetPodPath(filepath string) string - GenerateNodeID(hostName string, nvmeNQN string, fcWWNs []string, iscsiIQN string) (string, error) GetTopologyLabels(ctx context.Context, nodeName string) (map[string]string, error) + UpdateNodePortsAnnotation(ctx context.Context, nodeName string, iscsiIQN string, fcWWNs []string, nvmeNQN string) error IsBlock(devicePath string) (bool, error) GetFileSystemVolumeStats(path string) (VolumeStatistics, error) GetBlockVolumeStats(volumeId string) (VolumeStatistics, error) @@ -480,49 +483,6 @@ func (n NodeUtils) GetPodPath(origPath string) string { return path.Join(PrefixChrootOfHostRoot, origPath) } -func (n NodeUtils) GenerateNodeID(hostName string, nvmeNQN string, fcWWNs []string, iscsiIQN string) (string, error) { - var nodeId strings.Builder - nodeIdDelimiter := n.ConfigYaml.Parameters.Node_id_info.Delimiter - nodeIdFcDelimiter := n.ConfigYaml.Parameters.Node_id_info.Fcs_delimiter - nodeId.Grow(MaxNodeIdLength) - nodeId.WriteString(hostName) - nodeId.WriteString(nodeIdDelimiter) - - if len(nvmeNQN) > 0 { - if nodeId.Len()+len(nvmeNQN)+len(nodeIdDelimiter) <= MaxNodeIdLength { - nodeId.WriteString(nvmeNQN) - } else { - return "", fmt.Errorf(ErrorNoPortsCouldFitInNodeId, nodeId.String(), MaxNodeIdLength) - } - } - nodeId.WriteString(nodeIdDelimiter) - if len(fcWWNs) > 0 { - if nodeId.Len()+len(fcWWNs[0]) <= MaxNodeIdLength { - nodeId.WriteString(fcWWNs[0]) - } else if nvmeNQN == "" { - return "", fmt.Errorf(ErrorNoPortsCouldFitInNodeId, nodeId.String(), MaxNodeIdLength) - } - - for _, fcPort := range fcWWNs[1:] { - if nodeId.Len()+len(nodeIdFcDelimiter)+len(fcPort) <= MaxNodeIdLength { - nodeId.WriteString(nodeIdFcDelimiter) - nodeId.WriteString(fcPort) - } - } - } - if len(iscsiIQN) > 0 { - if nodeId.Len()+len(nodeIdDelimiter)+len(iscsiIQN) <= MaxNodeIdLength { - nodeId.WriteString(nodeIdDelimiter) - nodeId.WriteString(iscsiIQN) - } else if len(fcWWNs) == 0 && nvmeNQN == "" { - return "", fmt.Errorf(ErrorNoPortsCouldFitInNodeId, nodeId.String(), MaxNodeIdLength) - } - } - - finalNodeId := strings.TrimSuffix(nodeId.String(), ";") - return finalNodeId, nil -} - func (n NodeUtils) GetTopologyLabels(ctx context.Context, nodeName string) (map[string]string, error) { kubeConfig, err := rest.InClusterConfig() if err != nil { @@ -552,6 +512,73 @@ func (n NodeUtils) GetTopologyLabels(ctx context.Context, nodeName string) (map[ return topologyLabels, nil } +func (n NodeUtils) UpdateNodePortsAnnotation(ctx context.Context, nodeName string, + iscsiIQN string, fcWWNs []string, nvmeNQN string) error { + + kubeConfig, err := rest.InClusterConfig() + if err != nil { + logger.Infof("Failed to update initiators. Unable to load in-cluster configuration.") + return err + } + + client, err := kubernetes.NewForConfig(kubeConfig) + if err != nil { + logger.Infof("Failed to update initiators. Unable to create Kubernetes client.") + return err + } + + portsData := map[string]interface{}{ + "iscsi": []string{}, + "fc": []string{}, + "nvme": []string{}, + } + + if iscsiIQN != "" { + portsData["iscsi"] = []string{iscsiIQN} + } + + if len(fcWWNs) > 0 { + portsData["fc"] = fcWWNs + } + + if nvmeNQN != "" { + portsData["nvme"] = []string{nvmeNQN} + } + + jsonBytes, err := json.Marshal(portsData) + if err != nil { + return fmt.Errorf("marshal portsData: %w", err) + } + + patch := map[string]interface{}{ + "metadata": map[string]interface{}{ + "annotations": map[string]string{ + "block.csi.ibm.com/node-initiators": string(jsonBytes), + }, + }, + } + + patchBytes, err := json.Marshal(patch) + if err != nil { + return fmt.Errorf("marshal patch: %w", err) + } + + logger.Infof("Patching node %s with annotation block.csi.ibm.com/node-initiators", nodeName) + + _, err = client.CoreV1().Nodes().Patch(ctx, nodeName, + types.MergePatchType, patchBytes, patchOpts) + + if err != nil { + logger.Infof("failed PATCH node ports annotation.") + return err + } + + updatedNode, err := client.CoreV1().Nodes().Get(ctx, nodeName, getOpts) + logger.Infof("updatedNode: %s", updatedNode) + + return nil +} + func (n NodeUtils) IsBlock(devicePath string) (bool, error) { var stat unix.Stat_t err := unix.Stat(devicePath, &stat) diff --git a/node/pkg/driver/node_utils_test.go b/node/pkg/driver/node_utils_test.go index dd83444a2..2d474692e 100644 --- a/node/pkg/driver/node_utils_test.go +++ b/node/pkg/driver/node_utils_test.go @@ -306,102 +306,6 @@ func TestParseIscsiInitiators(t *testing.T) { } -func TestGenerateNodeID(t *testing.T) { - testCases := []struct { - name string - hostName string - nvmeNQN string - fcWWNs []string - iscsiIQN string - expErr error - expNodeId string - }{ - {name: "success all given, only nvme and fc in", - hostName: hostName, - nvmeNQN: nvmeNQN, - fcWWNs: fcWWNs, - iscsiIQN: iscsiIQN, - expNodeId: fmt.Sprintf("%s;%s;%s", hostName, nvmeNQN, strings.Join(fcWWNs, ":")), - }, - {name: "success only iscsi port", - hostName: hostName, - nvmeNQN: "", - fcWWNs: []string{}, - iscsiIQN: iscsiIQN, - expNodeId: fmt.Sprintf("%s;;;%s", hostName, iscsiIQN), - }, - {name: "success only fc ports", - hostName: hostName, - nvmeNQN: "", - fcWWNs: fcWWNs[:2], - iscsiIQN: "", - expNodeId: fmt.Sprintf("%s;;%s", hostName, strings.Join(fcWWNs[:2], ":")), - }, - {name: "success only nvme port", - hostName: hostName, - nvmeNQN: nvmeNQN, - fcWWNs: []string{}, - iscsiIQN: "", - expNodeId: fmt.Sprintf("%s;%s", hostName, nvmeNQN), - }, - {name: "success fc ports and iscsi port", - hostName: hostName, - nvmeNQN: "", - fcWWNs: fcWWNs[:2], - iscsiIQN: iscsiIQN, - expNodeId: fmt.Sprintf("%s;;%s;%s", hostName, strings.Join(fcWWNs[:2], ":"), iscsiIQN), - }, - {name: "fail long hostName on nvme port", - hostName: longHostName, - nvmeNQN: nvmeNQN, - fcWWNs: []string{}, - iscsiIQN: "", - expErr: errors.New(fmt.Sprintf("could not fit any ports in node id: %s;, length limit: %d", longHostName, maxNodeIdLength)), - }, - {name: "fail long hostName on fc ports", - hostName: longHostName, - nvmeNQN: "", - fcWWNs: fcWWNs[:2], - iscsiIQN: "", - expErr: errors.New(fmt.Sprintf("could not fit any ports in node id: %s;;, length limit: %d", longHostName, maxNodeIdLength)), - }, - {name: "fail long hostName on iscsi port", - hostName: longHostName, - nvmeNQN: "", - fcWWNs: []string{}, - iscsiIQN: iscsiIQN, - expErr: errors.New(fmt.Sprintf("could not fit any ports in node id: %s;;, length limit: %d", longHostName, maxNodeIdLength)), - }, - } - for _, tc := range testCases { - t.Run(tc.name, func(t *testing.T) { - - mockCtrl := gomock.NewController(t) - defer mockCtrl.Finish() - - fakeExecuter := mocks.NewMockExecuterInterface(mockCtrl) - nodeUtils := driver.NewNodeUtils(fakeExecuter, nil, ConfigYaml, nil) - - nodeId, err := nodeUtils.GenerateNodeID(tc.hostName, tc.nvmeNQN, tc.fcWWNs, tc.iscsiIQN) - - if tc.expErr != nil { - if err == nil || err.Error() != tc.expErr.Error() { - t.Fatalf("Expecting err: expected %v, got %v", tc.expErr, err) - } - - } else { - if err != nil { - t.Fatalf("err is not nil. got: %v", err) - } - if nodeId != tc.expNodeId { - t.Fatalf("wrong nodeId: expected %v, got %v", tc.expNodeId, nodeId) - } - - } - }) - } -} - func TestGetVolumeUuid(t *testing.T) { testCases := []struct { name string