CSI-5277/ separate addon code from controller server

controllers/scripts/csi_general/csi_pb2.sh

@@ -12,6 +12,7 @@ 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/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
 cd -
 

controllers/servers/csi/controller_server_manager.py

@@ -3,12 +3,11 @@
 from concurrent import futures
 
 import grpc
-from csi_general import csi_pb2_grpc, replication_pb2_grpc, volumegroup_pb2_grpc
+from csi_general import csi_pb2_grpc, volumegroup_pb2_grpc
 
 from controllers.common.config import config
 from controllers.common.csi_logger import get_stdout_logger
 from controllers.common.settings import CSI_CONTROLLER_SERVER_WORKERS
-from controllers.servers.csi.addons_server import ReplicationControllerServicer
 from controllers.servers.csi.csi_controller_server import CSIControllerServicer
 from controllers.servers.csi.volume_group_server import VolumeGroupControllerServicer
 
@@ -24,7 +23,6 @@ class ControllerServerManager:
     def __init__(self, array_endpoint):
         self.endpoint = array_endpoint
         self.csi_servicer = CSIControllerServicer()
-        self.replication_servicer = ReplicationControllerServicer()
         self.volume_group_servicer = VolumeGroupControllerServicer()
 
     def start_server(self):
@@ -33,7 +31,6 @@ def start_server(self):
 
         csi_pb2_grpc.add_ControllerServicer_to_server(self.csi_servicer, controller_server)
         csi_pb2_grpc.add_IdentityServicer_to_server(self.csi_servicer, controller_server)
-        replication_pb2_grpc.add_ControllerServicer_to_server(self.replication_servicer, controller_server)
         volumegroup_pb2_grpc.add_ControllerServicer_to_server(self.volume_group_servicer, controller_server)
 
         # bind the server to the port defined above

controllers/servers/csi/csi_addons_server/csi_addons_server_manager.py

@@ -0,0 +1,50 @@
+import os
+import time
+from concurrent import futures
+
+import grpc
+from csi_general import identity_pb2_grpc, replication_pb2_grpc
+
+from controllers.common.csi_logger import get_stdout_logger
+from controllers.common.settings import CSI_CONTROLLER_SERVER_WORKERS
+from controllers.servers.csi.csi_addons_server.replication_controller_servicer import ReplicationControllerServicer
+from controllers.servers.csi.csi_addons_server.identity_controller_servicer import IdentityControllerServicer
+
+logger = get_stdout_logger()
+
+
+def get_max_workers_count():
+    cpu_count = (os.cpu_count() or 1) + 4
+    return CSI_CONTROLLER_SERVER_WORKERS if cpu_count < CSI_CONTROLLER_SERVER_WORKERS else None
+
+
+class CSIAddonsServerManager:
+    def __init__(self, csi_addons_endpoint):
+        self.endpoint = csi_addons_endpoint
+        self.replication_servicer = ReplicationControllerServicer()
+        self.identity_servicer = IdentityControllerServicer()
+
+    def start_server(self):
+        max_workers = get_max_workers_count()
+        csi_addons_server = grpc.server(futures.ThreadPoolExecutor(max_workers=max_workers))
+
+        replication_pb2_grpc.add_ControllerServicer_to_server(self.replication_servicer, csi_addons_server)
+        identity_pb2_grpc.add_IdentityServicer_to_server(self.identity_servicer, csi_addons_server)
+
+        # bind the server to the port defined above
+        # controller_server.add_insecure_port('[::]:{}'.format(self.server_port))
+        # controller_server.add_insecure_port('unix://{}'.format(self.server_port))
+        csi_addons_server.add_insecure_port(self.endpoint)
+
+        # start the server
+        logger.debug("Listening for connections on endpoint address: {}".format(self.endpoint))
+
+        csi_addons_server.start()
+        logger.debug('CSI Addons Server running ...')
+
+        try:
+            while True:
+                time.sleep(60 * 60 * 60)
+        except KeyboardInterrupt:
+            csi_addons_server.stop(0)
+            logger.debug('CSI Addons Server Stopped ...')

controllers/servers/csi/csi_addons_server/identity_controller_servicer.py

@@ -0,0 +1,50 @@
+import grpc
+
+from csi_general import identity_pb2 as pb2
+from csi_general import identity_pb2_grpc as pb2_grpc
+
+from controllers.common.config import config as common_config
+from controllers.servers.csi.decorators import csi_method
+from controllers.servers.csi.exception_handler import build_error_response
+from controllers.common.csi_logger import get_stdout_logger
+
+logger = get_stdout_logger()
+
+
+class IdentityControllerServicer(pb2_grpc.IdentityServicer):
+
+    @csi_method(error_response_type=pb2.GetIdentityResponse)
+    def GetIdentity(self, request, context):
+        name = common_config.identity.name
+        version = common_config.identity.version
+
+        if not name or not version:
+            message = "plugin name or version cannot be empty"
+            return build_error_response(message, context, grpc.StatusCode.INTERNAL, pb2.GetIdentityResponse)
+
+        return pb2.GetIdentityResponse(name=name, vendor_version=version)
+
+    def GetCapabilities(self, request, context):
+        logger.info("GetCapabilities")
+        response = pb2.GetCapabilitiesResponse(
+            capabilities=[self._get_replication_capability(),
+                          self._get_controller_capability()])
+
+        logger.info("finished GetCapabilities")
+        return response
+
+    def _get_replication_capability(self):
+        types = pb2.Capability.VolumeReplication.Type
+        capability_enum_value = types.Value("VOLUME_REPLICATION")
+        return pb2.Capability(
+            volume_replication=pb2.Capability.VolumeReplication(type=capability_enum_value))
+
+    def _get_controller_capability(self):
+        types = pb2.Capability.Service.Type
+        capability_enum_value = types.Value("CONTROLLER_SERVICE")
+        return pb2.Capability(
+            service=pb2.Capability.Service(type=capability_enum_value))
+
+    def Probe(self, request, context):
+        context.set_code(grpc.StatusCode.OK)
+        return pb2.ProbeResponse()

controllers/servers/csi/main.py

@@ -1,19 +1,32 @@
 from argparse import ArgumentParser
+from threading import Thread
 
 from controllers.common.csi_logger import set_log_level
 from controllers.servers.csi.controller_server_manager import ControllerServerManager
+from controllers.servers.csi.csi_addons_server.csi_addons_server_manager import CSIAddonsServerManager
 
 
 def main():
     parser = ArgumentParser()
     parser.add_argument("-e", "--csi-endpoint", dest="endpoint", help="grpc endpoint")
+    parser.add_argument("-a", "--csi-addons-endpoint", dest="addonsendpoint", help="CSI-Addons grpc endpoint")
     parser.add_argument("-l", "--loglevel", dest="loglevel", help="log level")
     arguments = parser.parse_args()
 
     set_log_level(arguments.loglevel)
 
-    server_manager = ControllerServerManager(arguments.endpoint)
-    server_manager.start_server()
+    csi_controller_server_manager = ControllerServerManager(arguments.endpoint)
+    csi_addons_server_manager = CSIAddonsServerManager(arguments.addonsendpoint)
+    _start_servers(csi_controller_server_manager, csi_addons_server_manager)
+
+
+def _start_servers(csi_controller_server_manager, csi_addons_server_manager):
+    servers = (
+        csi_controller_server_manager.start_server,
+        csi_addons_server_manager.start_server)
+    for server_function in servers:
+        thread = Thread(target=server_function,)
+        thread.start()
 
 
 if __name__ == '__main__':

controllers/tests/controller_server/addons_server_test.py

@@ -7,7 +7,7 @@
 from controllers.servers.settings import PARAMETERS_SYSTEM_ID, PARAMETERS_COPY_TYPE, PARAMETERS_REPLICATION_POLICY
 from controllers.array_action.settings import REPLICATION_TYPE_MIRROR, REPLICATION_TYPE_EAR, REPLICATION_COPY_TYPE_SYNC
 from controllers.array_action.array_action_types import ReplicationRequest
-from controllers.servers.csi.addons_server import ReplicationControllerServicer
+from controllers.servers.csi.csi_addons_server.replication_controller_servicer import ReplicationControllerServicer
 from controllers.tests import utils
 from controllers.tests.common.test_settings import VOLUME_NAME, VOLUME_UID, OBJECT_INTERNAL_ID, \
     OTHER_OBJECT_INTERNAL_ID, REPLICATION_NAME, SYSTEM_ID, COPY_TYPE, SECRET_USERNAME_VALUE, SECRET_PASSWORD_VALUE, \
@@ -16,7 +16,7 @@
 from controllers.tests.controller_server.csi_controller_server_test import (CommonControllerTest)
 from controllers.tests.utils import ProtoBufMock
 
-ADDON_SERVER_PATH = "controllers.servers.csi.addons_server"
+ADDON_SERVER_PATH = "controllers.servers.csi.csi_addons_server.replication_controller_servicer"
 
 
 class BaseReplicationSetUp(unittest.TestCase):