EOS-26899: MessageServer: message server class refactor (Seagate#677)

* Fix init phase of Miniprovisioning (Seagate#645) * Fix init phase of Miniprovisioning Signed-off-by: Selvakumar <[email protected]> * Fix test cases related to message_bus Signed-off-by: Selvakumar <[email protected]> * Remove creation of config files Signed-off-by: Selvakumar <[email protected]> * Make use of components logger (Seagate#650) * Make use of components logger Signed-off-by: Selvakumar <[email protected]> * Use utils logger if initialised Signed-off-by: Selvakumar <[email protected]> * Ignore logging in message_bus if logger is not initialised (Seagate#652) * Ignore logging if logger is not initialised Signed-off-by: Selvakumar <[email protected]> * Ignore logging if logger is not initialised Signed-off-by: Selvakumar <[email protected]> * Updated message_bus tests to take cluster.conf path as input and get kafka endpoints from conf. (Seagate#658) * PATCH: update PyYaml version Signed-off-by: Rohit Dwivedi <[email protected]> * EOS-25893: Implement init method for Event_message (Seagate#670) * Iem prep and messagebus init changes Signed-off-by: suryakumar.kumaravelan <[email protected]> * Update event_message.py * Update event_message.py * Update event_message.py * Update test_iem.py * Update event_message.py * Update test_iem.py * Update event_message.py Co-authored-by: Sachin Punadikar <[email protected]> Co-authored-by: Selva Nambi <[email protected]> * Message bus server no conf patch. updated iem and message bus server py files Signed-off-by: suryakumar.kumaravelan <[email protected]> * code optimize Signed-off-by: suryakumar.kumaravelan <[email protected]> * code optimize Signed-off-by: suryakumar.kumaravelan <[email protected]> Co-authored-by: Selva Nambi <[email protected]> Co-authored-by: veerendra-simha-garikipati <[email protected]> Co-authored-by: Rohit Dwivedi <[email protected]> Co-authored-by: Sachin Punadikar <[email protected]> Signed-off-by: suryakumar.kumaravelan <[email protected]>
suryakumar1024 · Mar 21, 2022 · 2c6b191 · 2c6b191
1 parent 0a776db
commit 2c6b191
Show file tree

Hide file tree

Showing 5 changed files with 44 additions and 20 deletions.
diff --git a/py-utils/src/utils/audit_log/audit_log_server.py b/py-utils/src/utils/audit_log/audit_log_server.py
@@ -18,7 +18,7 @@
 import json
 from aiohttp import web
 
-from cortx.utils.utils_server import RestServer
+from cortx.utils.utils_server import MessageServer
 from cortx.utils.audit_log.error import AuditLogError
 from cortx.utils.utils_server.error import RestServerError
 from cortx.utils.message_bus import MessageProducer
@@ -29,7 +29,7 @@
     message_type='audit_messages', method='sync')
 
 
-class AuditLogRequestHandler(RestServer):
+class AuditLogRequestHandler(MessageServer):
     """Rest interface of Audit log."""
     # webhook_info will be removed once webhook_info store to persistent storage
     webhook_info = None

diff --git a/py-utils/src/utils/iem_framework/iem_server.py b/py-utils/src/utils/iem_framework/iem_server.py
@@ -17,7 +17,7 @@
 
 import json
 from aiohttp import web
-from cortx.utils.utils_server import RestServer
+from cortx.utils.utils_server import MessageServer
 from cortx.utils.iem_framework import EventMessage
 from cortx.utils.utils_server.error import RestServerError
 from cortx.utils.iem_framework.error import EventMessageError
@@ -27,7 +27,7 @@
 routes = web.RouteTableDef()
 
 
-class IemRequestHandler(RestServer):
+class IemRequestHandler(MessageServer):
     """ Rest interface of Iem """
 
     @staticmethod

diff --git a/py-utils/src/utils/message_bus/message_bus_server.py b/py-utils/src/utils/message_bus/message_bus_server.py
@@ -18,7 +18,7 @@
 import json
 from aiohttp import web
 
-from cortx.utils.utils_server import RestServer
+from cortx.utils.utils_server import MessageServer
 from cortx.utils.message_bus.error import MessageBusError
 from cortx.utils.utils_server.error import RestServerError
 from cortx.utils.message_bus import MessageConsumer, MessageProducer, MessageBus
@@ -29,7 +29,7 @@
 routes = web.RouteTableDef()
 
 
-class MessageBusRequestHandler(RestServer):
+class MessageBusRequestHandler(MessageServer):
     """ Rest interface of message bus """
 
     @staticmethod

diff --git a/py-utils/src/utils/utils_server/__init__.py b/py-utils/src/utils/utils_server/__init__.py
@@ -17,6 +17,6 @@
 
 __title__ = 'utils_server'
 
-from cortx.utils.utils_server.utils_server import RestServer
+from cortx.utils.utils_server.utils_server import MessageServer
 from cortx.utils.utils_server.error import RestServerError
 
diff --git a/py-utils/src/utils/utils_server/utils_server.py b/py-utils/src/utils/utils_server/utils_server.py
@@ -16,21 +16,42 @@
 # please email [email protected] or [email protected].
 
 import argparse
+import errno
 from argparse import RawTextHelpFormatter
 from aiohttp import web
+
 from cortx.utils.log import Log
 from cortx.utils.common import CortxConf
+from cortx.utils.conf_store import Conf
+from cortx.utils.errors import UtilsError
+from cortx.utils.message_bus import MessageBus
+
 
+class UtilsServerError(UtilsError):
+    """UtilsServerError exception with error code and output."""
+    def __init__(self, rc, message, *args):
+        super().__init__(rc, message, *args)
 
-class RestServer:
-    """ Base class for Cortx Rest Server implementation """
 
+class UtilsServer:
+    """Base class for Cortx Rest Server implementation."""
     def __init__(self):
-        app = web.Application()
+        self.app = web.Application()
+
+    def run_app(self, web):
+       Log.info("Starting Message Server 0.0.0.0 on port 28300")
+       web.run_app(self.app, port=28300)
+
+
+class MessageServer(UtilsServer):
+    """Base class for Cortx Rest Server implementation."""
+    def __init__(self, message_server_endpoints):
+        super().__init__()
+        MessageBus.init(message_server_endpoints=message_server_endpoints)
         from cortx.utils.iem_framework import IemRequestHandler
         from cortx.utils.message_bus import MessageBusRequestHandler
         from cortx.utils.audit_log import AuditLogRequestHandler
-        app.add_routes([web.post('/EventMessage/event', IemRequestHandler.send), \
+        self.app.add_routes([web.post('/EventMessage/event', IemRequestHandler.send), \
             web.get('/EventMessage/event', IemRequestHandler.receive), \
             web.post('/MessageBus/message/{message_type}', \
             MessageBusRequestHandler.send), \
@@ -41,28 +62,31 @@ def __init__(self):
             web.post('/AuditLog/webhook/', \
             AuditLogRequestHandler.send_webhook_info)
             ])
-
-        Log.info("Starting Message Server 0.0.0.0 on port 28300")
-        web.run_app(app, port=28300)
+        super().run_app(web)
 
 
 if __name__ == '__main__':
-    import os
     from cortx.utils.conf_store import Conf
     parser = argparse.ArgumentParser(description='Utils server CLI',
         formatter_class=RawTextHelpFormatter)
     parser.add_argument('-c', '--config', dest='cluster_conf',\
         help="Cluster config file path for Support Bundle",\
         default='yaml:///etc/cortx/cluster.conf')
     args=parser.parse_args()
-
-    CortxConf.init(cluster_conf=args.cluster_conf)
+    cluster_conf = args.cluster_conf
+    Conf.load('config', cluster_conf, skip_reload=True)
+    CortxConf.init(cluster_conf=cluster_conf)
     # Get the log path
-    log_dir = CortxConf.get('log_dir', '/var/log')
+    log_dir = CortxConf.get_storage_path('log')
+    if not log_dir:
+        raise UtilsServerError(errno.EINVAL, "Fail to initialize logger."+\
+            " Unable to find log_dir path entry")
     utils_log_path = CortxConf.get_log_path('utils_server', base_dir=log_dir)
     # Get the log level
     log_level = CortxConf.get('utils>log_level', 'INFO')
-
     Log.init('utils_server', utils_log_path, level=log_level, backup_count=5, \
         file_size_in_mb=5)
-    RestServer()
+    message_bus_backend = Conf.get('config', 'cortx>utils>message_bus_backend')
+    message_server_endpoints = Conf.get('config',\
+            f'cortx>external>{message_bus_backend}>endpoints')
+    MessageServer(message_server_endpoints)