Feature csi 5536 add fence implementation

controllers/array_action/array_mediator_svc.py

@@ -1,29 +1,30 @@
 from collections import defaultdict
+from datetime import datetime, timedelta
 from io import StringIO
 from random import choice
-from datetime import datetime, timedelta
 
 from packaging.version import Version
 from pysvc import errors as svc_errors
 from pysvc.unified.client import connect
 from pysvc.unified.response import CLIFailureError, SVCResponse
 from retry import retry
 
-from controllers.common.config import config
 import controllers.array_action.errors as array_errors
 import controllers.array_action.settings as array_settings
-from controllers.array_action.registration_cache import SVC_REGISTRATION_CACHE
-from controllers.array_action import svc_messages
 import controllers.servers.settings as controller_settings
-from controllers.servers.csi.decorators import register_csi_plugin
+from controllers.array_action import svc_messages
 from controllers.array_action.array_action_types import Volume, Snapshot, Replication, Host, VolumeGroup, ThinVolume
 from controllers.array_action.array_mediator_abstract import ArrayMediatorAbstract
+from controllers.array_action.fence_interface import FenceInterface
+from controllers.array_action.registration_cache import SVC_REGISTRATION_CACHE
 from controllers.array_action.utils import ClassProperty, convert_scsi_id_to_nguid
 from controllers.array_action.volume_group_interface import VolumeGroupInterface
 from controllers.common import settings as common_settings
+from controllers.common.config import config
 from controllers.common.csi_logger import get_stdout_logger
-from controllers.servers.utils import get_connectivity_type_ports, split_string, is_call_home_enabled
+from controllers.servers.csi.decorators import register_csi_plugin
 from controllers.servers.settings import UNIQUE_KEY_KEY
+from controllers.servers.utils import get_connectivity_type_ports, split_string, is_call_home_enabled
 
 array_connections_dict = {}
 logger = get_stdout_logger()
@@ -104,7 +105,7 @@ def _get_space_efficiency_kwargs(space_efficiency):
 
 def _is_space_efficiency_matches_source(parameter_space_efficiency, array_space_efficiency):
     return (not parameter_space_efficiency and array_space_efficiency == common_settings.SPACE_EFFICIENCY_THICK) or \
-           (parameter_space_efficiency and parameter_space_efficiency == array_space_efficiency)
+        (parameter_space_efficiency and parameter_space_efficiency == array_space_efficiency)
 
 
 def build_create_volume_in_volume_group_kwargs(pool, io_group, source_id):
@@ -224,7 +225,7 @@ def _get_cli_volume_space_efficiency_aliases(cli_volume):
     return space_efficiency_aliases
 
 
-class SVCArrayMediator(ArrayMediatorAbstract, VolumeGroupInterface):
+class SVCArrayMediator(ArrayMediatorAbstract, VolumeGroupInterface, FenceInterface):
     ARRAY_ACTIONS = {}
     BLOCK_SIZE_IN_BYTES = 512
     MAX_LUN_NUMBER = 511
@@ -888,9 +889,12 @@ def _create_snapshot(self, target_volume_name, source_cli_volume, space_efficien
             self._rollback_create_snapshot(target_volume_name)
             raise ex
 
+    def _lsmdiskgrp(self, **kwargs):
+        return self.client.svcinfo.lsmdiskgrp(**kwargs)
+
     def _get_pool_site(self, pool):
         filter_value = 'name={}'.format(pool)
-        cli_pool = self.client.svcinfo.lsmdiskgrp(filtervalue=filter_value).as_single_element
+        cli_pool = self._lsmdiskgrp(filtervalue=filter_value).as_single_element
         if cli_pool:
             return cli_pool.site_name
         raise array_errors.PoolDoesNotExist(pool, self.endpoint)
@@ -2064,7 +2068,58 @@ def remove_volume_from_volume_group(self, volume_id):
         cli_volume = self._get_cli_volume_by_wwn(volume_id, not_exist_err=True)
         self._change_volume_group(cli_volume.id, None)
 
-    def register_plugin(self, unique_key,  metadata):
+    def _get_ownership_group_pools(self, ownership_group):
+        logger.info(svc_messages.GET_OWNERSHIP_GROUP_POOLS.format(ownership_group))
+        filter_value = 'owner_name={}'.format(ownership_group)
+        cli_pools = self._lsmdiskgrp(filtervalue=filter_value).as_list
+        return cli_pools
+
+    def is_fenced(self, fence_ownership_group):
+        ownership_group_pools = self._get_ownership_group_pools(fence_ownership_group)
+        if len(ownership_group_pools) == 0:
+            logger.info(svc_messages.NO_POOLS_FOUND_IN_OWNERSHIP_GROUP.format(fence_ownership_group))
+            return True
+
+        logger.info(svc_messages.POOLS_FOUND_IN_OWNERSHIP_GROUP.format(fence_ownership_group, ownership_group_pools))
+        return False
+
+    def _chmdiskgrp(self, pool_id, **cli_kwargs):
+        self.client.svctask.chmdiskgrp(object_id=pool_id, **cli_kwargs)
+
+    def _remove_all_mappings_from_ownership_group(self, ownership_group):
+        logger.info(svc_messages.REMOVING_ALL_MAPPINGS_FROM_OWNERSHIP_GROUP.format(ownership_group))
+        filter_value = 'owner_name={}'.format(ownership_group)
+
+        hosts = self.client.svcinfo.lshost(filtervalue=filter_value).as_list
+        host_names = [host.name for host in hosts]
+
+        volumes = self._lsvdisk_list(filtervalue=filter_value)
+        volume_names = [volume.name for volume in volumes]
+
+        mappings = self.client.svcinfo.lshostvdiskmap().as_list
+
+        relevant_mappings = [mapping for mapping in mappings if
+                             mapping.name in host_names and mapping.vdisk_name in volume_names]
+        logger.info(svc_messages.REMOVING_MAPPINGS.format(relevant_mappings))
+        for mapping in relevant_mappings:
+            self.client.svctask.rmvdiskhostmap(vdisk_name=mapping.vdisk_name, host=mapping.name)
+
+    def _change_pools_ownership_group(self, ownership_group, pools):
+        logger.info(svc_messages.CHANGE_POOLS_OWNERSHIP_GROUP.format(ownership_group))
+        for pool in pools:
+            self._chmdiskgrp(pool.id, ownershipgroup=ownership_group)
+
+    def fence(self, fence_ownership_group, unfence_ownership_group):
+        ownership_group_pools = self._get_ownership_group_pools(fence_ownership_group)
+        if len(ownership_group_pools) == 0:
+            logger.info(svc_messages.NO_POOLS_FOUND_IN_OWNERSHIP_GROUP.format(fence_ownership_group))
+            return
+
+        self._remove_all_mappings_from_ownership_group(fence_ownership_group)
+
+        self._change_pools_ownership_group(unfence_ownership_group, ownership_group_pools)
+
+    def register_plugin(self, unique_key, metadata):
         if is_call_home_enabled() and self._is_registerplugin_supported() and \
                 self._is_plugin_needs_to_be_registered(unique_key):
             self._register_plugin(unique_key, metadata)
@@ -2102,4 +2157,4 @@ def _registerplugin(self, unique_key, metadata):
         except Exception as ex:
             logger.error("exception encountered during"
                          "registering {} plugin using {} unique key with [{}] metadata: {}".format(
-                             array_settings.REGISTRATION_PLUGIN, unique_key, metadata, ex))
+                array_settings.REGISTRATION_PLUGIN, unique_key, metadata, ex))

controllers/array_action/fence_interface.py

@@ -0,0 +1,37 @@
+from abc import ABC, abstractmethod
+
+
+class FenceInterface(ABC):
+
+    @abstractmethod
+    def is_fenced(self, fence_ownership_group):
+        """
+        This function should check if the fence_ownership_group is already fenced (no pools in the og)
+
+        Args:
+            fence_ownership_group : name of the ownership group that should be fenced
+
+        Returns:
+            bool
+
+        Raises:
+            None
+        """
+        raise NotImplementedError
+
+    @abstractmethod
+    def fence(self, fence_ownership_group, unfence_ownership_group):
+        """
+        This function should fence the fence_ownership_group and unfence the unfence_ownership_group
+
+        Args:
+            fence_ownership_group : name of the ownership group that should be fenced
+            unfence_ownership_group : name of the ownership group that should be unfenced
+
+        Returns:
+            None
+
+        Raises:
+            None
+        """
+        raise NotImplementedError

controllers/array_action/svc_messages.py

@@ -10,3 +10,9 @@
 CREATE_HOST_WITHOUT_IO_GROUP = 'Created host {} with port {}'
 CREATE_HOST_WITH_IO_GROUP = 'Created host {} with port [{}] and with io_group [{}]'
 CHANGE_HOST_PROTOCOL = 'Changed host {} protocol to: {}'
+GET_OWNERSHIP_GROUP_POOLS = 'Getting pools for ownership group {}'
+NO_POOLS_FOUND_IN_OWNERSHIP_GROUP = 'No pools found in ownership group {}'
+POOLS_FOUND_IN_OWNERSHIP_GROUP = 'Pools found in ownership group {}: {}'
+REMOVING_ALL_MAPPINGS_FROM_OWNERSHIP_GROUP = 'Removing all mappings from ownership group {}'
+REMOVING_MAPPINGS = 'Removing mappings {}'
+CHANGE_POOLS_OWNERSHIP_GROUP = 'Changing pools ownership group to {}'

controllers/scripts/csi_general/csi_pb2.sh

@@ -12,8 +12,10 @@ cd ./proto/${PB2_DIR}
 curl -O https://raw.githubusercontent.com/container-storage-interface/spec/${CSI_VERSION}/csi.proto
 curl -O https://raw.githubusercontent.com/IBM/csi-volume-group/${VG_VERSION}/volumegroup/volumegroup.proto
 curl -O https://raw.githubusercontent.com/csi-addons/spec/v0.2.0/replication/replication.proto
+curl -O https://raw.githubusercontent.com/csi-addons/spec/v0.2.0/fence/fence.proto
 curl -O https://raw.githubusercontent.com/csi-addons/spec/main/identity/identity.proto
 sed -i 's|github.com/container-storage-interface/spec/lib/go/csi/csi.proto|csi_general/csi.proto|g' replication.proto
+sed -i 's|github.com/container-storage-interface/spec/lib/go/csi/csi.proto|csi_general/csi.proto|g' fence.proto
 cd -
 
 python -m grpc_tools.protoc --proto_path=proto \

controllers/servers/csi/csi_addons_server/fence_controller_servicer.py

@@ -0,0 +1,47 @@
+from csi_general import fence_pb2_grpc, fence_pb2
+
+from controllers.array_action.storage_agent import get_agent, detect_array_type
+from controllers.common.csi_logger import get_stdout_logger
+from controllers.servers import utils
+from controllers.servers.csi.decorators import csi_fence_method
+
+logger = get_stdout_logger()
+
+
+def _is_already_handled(mediator, fence_ownership_group):
+    return mediator.is_fenced(fence_ownership_group)
+
+
+def _fence_cluster_network(mediator, fence_ownership_group, unfence_ownership_group):
+    logger.info("fencing {}".format(fence_ownership_group))
+    mediator.fence(fence_ownership_group, unfence_ownership_group)
+
+
+def handle_fencing(request):
+    utils.validate_fencing_request(request)
+    fence_ownership_group = request.parameters["fenceToken"]
+    unfence_ownership_group = request.parameters["unfenceToken"]
+
+    connection_info = utils.get_array_connection_info_from_secrets(request.secrets)
+    array_type = detect_array_type(connection_info.array_addresses)
+    with get_agent(connection_info, array_type).get_mediator() as mediator:
+        # idempotence - check if the fence_ownership_group is already fenced (no pools in the og)
+        if _is_already_handled(mediator, fence_ownership_group):
+            logger.info("{} is fenced".format(fence_ownership_group))
+            return fence_pb2.FenceClusterNetworkResponse()
+        _fence_cluster_network(mediator, fence_ownership_group, unfence_ownership_group)
+        return fence_pb2.FenceClusterNetworkResponse()
+
+
+class FenceControllerServicer(fence_pb2_grpc.FenceControllerServicer):
+    @csi_fence_method(error_response_type=fence_pb2.FenceClusterNetworkResponse)
+    def FenceClusterNetwork(self, request, context):
+        return handle_fencing(request)
+
+    @csi_fence_method(error_response_type=fence_pb2.UnfenceClusterNetworkResponse)
+    def UnfenceClusterNetwork(self, request, context):
+        return handle_fencing(request)
+
+    @csi_fence_method(error_response_type=fence_pb2.ListClusterFenceResponse)
+    def ListClusterFence(self, request, context):
+        raise NotImplementedError()

controllers/servers/csi/csi_addons_server/identity_controller_servicer.py

@@ -28,7 +28,8 @@ def GetCapabilities(self, request, context):
         logger.info("GetCapabilities")
         response = pb2.GetCapabilitiesResponse(
             capabilities=[self._get_replication_capability(),
-                          self._get_controller_capability()])
+                          self._get_controller_capability(),
+                          self._get_network_fence_capability()])
 
         logger.info("finished GetCapabilities")
         return response
@@ -45,6 +46,12 @@ def _get_controller_capability(self):
         return pb2.Capability(
             service=pb2.Capability.Service(type=capability_enum_value))
 
+    def _get_network_fence_capability(self):
+        types = pb2.Capability.NetworkFence.Type
+        capability_enum_value = types.Value("NETWORK_FENCE")
+        return pb2.Capability(
+            network_fence=pb2.Capability.NetworkFence(type=capability_enum_value))
+
     def Probe(self, request, context):
         context.set_code(grpc.StatusCode.OK)
         return pb2.ProbeResponse()

controllers/servers/csi/decorators.py

@@ -24,6 +24,16 @@ def call_csi_method(controller_method, servicer, request, context):
     return call_csi_method
 
 
+def csi_fence_method(error_response_type):
+    @decorator
+    def call_csi_method(controller_method, servicer, request, context):
+        lock_id = request.parameters.get('fenceToken', '')
+        return _set_sync_lock(lock_id, 'fenceToken', error_response_type,
+                              controller_method, servicer, request, context)
+
+    return call_csi_method
+
+
 def csi_replication_method(error_response_type):
     @decorator
     def call_csi_method(controller_method, servicer, request, context):

controllers/servers/csi/main.py

@@ -1,19 +1,19 @@
 import os
 from argparse import ArgumentParser
-from threading import Thread
-from concurrent import futures
-import grpc
 from concurrent import futures
+from threading import Thread
 
-from csi_general import csi_pb2_grpc, volumegroup_pb2_grpc, identity_pb2_grpc, replication_pb2_grpc
+import grpc
+from csi_general import csi_pb2_grpc, volumegroup_pb2_grpc, identity_pb2_grpc, replication_pb2_grpc, fence_pb2_grpc
 
 from controllers.common.csi_logger import set_log_level
 from controllers.common.settings import CSI_CONTROLLER_SERVER_WORKERS
-from controllers.servers.csi.server_manager import ServerManager
 from controllers.servers.csi.controller_server.csi_controller_server import CSIControllerServicer
 from controllers.servers.csi.controller_server.volume_group_server import VolumeGroupControllerServicer
-from controllers.servers.csi.csi_addons_server.replication_controller_servicer import ReplicationControllerServicer
+from controllers.servers.csi.csi_addons_server.fence_controller_servicer import FenceControllerServicer
 from controllers.servers.csi.csi_addons_server.identity_controller_servicer import IdentityControllerServicer
+from controllers.servers.csi.csi_addons_server.replication_controller_servicer import ReplicationControllerServicer
+from controllers.servers.csi.server_manager import ServerManager
 
 
 def main():
@@ -56,8 +56,10 @@ def _add_csi_controller_servicers(controller_server):
 def _add_csi_addons_servicers(csi_addons_server):
     replication_servicer = ReplicationControllerServicer()
     identity_servicer = IdentityControllerServicer()
+    fence_servicer = FenceControllerServicer()
     replication_pb2_grpc.add_ControllerServicer_to_server(replication_servicer, csi_addons_server)
     identity_pb2_grpc.add_IdentityServicer_to_server(identity_servicer, csi_addons_server)
+    fence_pb2_grpc.add_FenceControllerServicer_to_server(fence_servicer, csi_addons_server)
     return csi_addons_server
 
 
@@ -66,7 +68,7 @@ def _start_servers(csi_controller_server_manager, csi_addons_server_manager):
         csi_controller_server_manager.start_server,
         csi_addons_server_manager.start_server)
     for server_function in servers:
-        thread = Thread(target=server_function,)
+        thread = Thread(target=server_function, )
         thread.start()
 
 

controllers/servers/utils.py

@@ -1,8 +1,8 @@
-from os import getenv
 import json
 import re
 from hashlib import sha256
 from operator import eq
+from os import getenv
 
 import base58
 from csi_general import csi_pb2, volumegroup_pb2
@@ -278,15 +278,15 @@ def _validate_object_id(object_id, object_type=servers_settings.VOLUME_TYPE_NAME
         raise ValidationException(messages.WRONG_FORMAT_MESSAGE.format("volume id"))
 
 
-def _validate_request_required_field(field_value, field_name):
+def _validate_required_field(field_value, field_name):
     logger.debug("validating request {}".format(field_name))
     if not field_value:
         raise ValidationException(messages.PARAMETER_SHOULD_NOT_BE_EMPTY_MESSAGE.format(field_name))
 
 
 def _validate_minimum_request_fields(request, required_field_names):
     for required_field_name in required_field_names:
-        _validate_request_required_field(getattr(request, required_field_name), required_field_name)
+        _validate_required_field(getattr(request, required_field_name), required_field_name)
     validate_secrets(request.secrets)
 
 
@@ -851,3 +851,17 @@ def get_replication_object_type_and_id_info(request):
 
 def is_call_home_enabled():
     return getenv(servers_settings.ENABLE_CALL_HOME_ENV_VAR, 'true') == 'true'
+
+
+def _validate_parameters_fields(parameters, required_parameters_names):
+    for required_field_name in required_parameters_names:
+        _validate_required_field(parameters.get(required_field_name), required_field_name)
+
+
+def validate_fencing_request(request):
+    logger.debug("validating fencing request")
+
+    _validate_parameters_fields(request.parameters, ["fenceToken", "unfenceToken"])
+    validate_secrets(request.secrets)
+
+    logger.debug("fencing validation finished")