Merge pull request #57 from meaningfy-ws/feature/TED-180

Feature/ted 180

Makefile

@@ -182,6 +182,10 @@ staging-dotenv-file: guard-VAULT_ADDR guard-VAULT_TOKEN vault-installed
 	@ echo ENVIRONMENT=staging >> .env
 	@ echo SUBDOMAIN=staging. >> .env
 	@ echo RML_MAPPER_PATH=${RML_MAPPER_PATH} >> .env
+	@ echo ELK_HOST=localhost >> .env
+	@ echo ELK_PORT=5959 >> .env
+	@ echo ELK_VERSION=1 >> .env
+	@ echo LOGGING_TYPE=PY,ELK >> .env
 	@ echo XML_PROCESSOR_PATH=${XML_PROCESSOR_PATH} >> .env
 	@ echo AIRFLOW_INFRA_FOLDER=~/airflow-infra/staging >> .env
 	@ vault kv get -format="json" ted-staging/airflow | jq -r ".data.data | keys[] as \$$k | \"\(\$$k)=\(.[\$$k])\"" >> .env
@@ -195,6 +199,10 @@ dev-dotenv-file: guard-VAULT_ADDR guard-VAULT_TOKEN vault-installed
 	@ echo ENVIRONMENT=dev >> .env
 	@ echo SUBDOMAIN= >> .env
 	@ echo RML_MAPPER_PATH=${RML_MAPPER_PATH} >> .env
+	@ echo ELK_HOST=localhost >> .env
+	@ echo ELK_PORT=5959 >> .env
+	@ echo ELK_VERSION=1 >> .env
+	@ echo LOGGING_TYPE=PY,ELK >> .env
 	@ echo XML_PROCESSOR_PATH=${XML_PROCESSOR_PATH} >> .env
 	@ echo AIRFLOW_INFRA_FOLDER=${AIRFLOW_INFRA_FOLDER} >> .env
 	@ vault kv get -format="json" ted-dev/airflow | jq -r ".data.data | keys[] as \$$k | \"\(\$$k)=\(.[\$$k])\"" >> .env
@@ -209,20 +217,39 @@ prod-dotenv-file: guard-VAULT_ADDR guard-VAULT_TOKEN vault-installed
 	@ echo ENVIRONMENT=prod >> .env
 	@ echo SUBDOMAIN= >> .env
 	@ echo RML_MAPPER_PATH=${RML_MAPPER_PATH} >> .env
+	@ echo ELK_HOST=localhost >> .env
+	@ echo ELK_PORT=5959 >> .env
+	@ echo ELK_VERSION=1 >> .env
+	@ echo LOGGING_TYPE=PY,ELK >> .env
 	@ echo XML_PROCESSOR_PATH=${XML_PROCESSOR_PATH} >> .env
 	@ echo AIRFLOW_INFRA_FOLDER=~/airflow-infra/prod >> .env
 	@ vault kv get -format="json" ted-prod/airflow | jq -r ".data.data | keys[] as \$$k | \"\(\$$k)=\(.[\$$k])\"" >> .env
 	@ vault kv get -format="json" ted-prod/mongo-db | jq -r ".data.data | keys[] as \$$k | \"\(\$$k)=\(.[\$$k])\"" >> .env
 
-local-dotenv-file: rml-mapper-path-add-dotenv-file
+local-dotenv-file: rml-mapper-path-add-dotenv-file elk-add-dotenv-file logging-add-dotenv-file
 
 rml-mapper-path-add-dotenv-file:
 	@ echo -e "$(BUILD_PRINT)Add rml-mapper path to local .env file $(END_BUILD_PRINT)"
 	@ sed -i '/^RML_MAPPER_PATH/d' .env
 	@ echo RML_MAPPER_PATH=${RML_MAPPER_PATH} >> .env
 
+elk-add-dotenv-file:
+	@ echo -e "$(BUILD_PRINT)Add elk config to local .env file $(END_BUILD_PRINT)"
+	@ sed -i '/^ELK_HOST/d' .env
+	@ echo ELK_HOST=localhost >> .env
+	@ sed -i '/^ELK_PORT/d' .env
+	@ echo ELK_PORT=5959 >> .env
+	@ sed -i '/^ELK_VERSION/d' .env
+	@ echo ELK_VERSION=1 >> .env
+
+logging-add-dotenv-file:
+	@ echo -e "$(BUILD_PRINT)Add logging config to local .env file $(END_BUILD_PRINT)"
+	@ sed -i '/^LOGGING_TYPE/d' .env
+	@ echo LOGGING_TYPE=PY,ELK >> .env
+
 refresh-normaliser-mapping-files:
 	@ python -m ted_sws.metadata_normaliser.entrypoints.generate_mapping_resources
+
 #clean-mongo-db:
 #	@ export PYTHONPATH=$(PWD) && python ./tests/clean_mongo_db.py
 

requirements.txt

@@ -9,5 +9,8 @@ apache-airflow==2.2.5
 hvac==0.11.2
 SPARQLWrapper==1.8.5
 pandas==1.3.5
+python-logstash~=0.4.8
+pymessagebus~=1.2
 click~=8.1.0
-openpyxl==3.0.9
+openpyxl==3.0.9
+

setup.py

@@ -75,7 +75,6 @@ def open_local(paths, mode="r", encoding="utf8"):
     packages=packages,
     entry_points={
         "console_scripts": [
-            # "rdfpipe = rdflib.tools.rdfpipe:main", # inspired form rdflib, replace as needed
             "transformer = ted_sws.notice_transformer.entrypoints.cmd_mapping_suite_transformer:main",
             "normalisation_resource_generator = ted_sws.metadata_normaliser.entrypoints.generate_mapping_resources:main"
         ],

ted_sws/__init__.py

@@ -12,6 +12,7 @@
 import os
 
 import dotenv
+
 from ted_sws.core.adapters.config_resolver import VaultAndEnvConfigResolver
 from ted_sws.core.adapters.vault_secrets_store import VaultSecretsStore
 
@@ -64,14 +65,36 @@ def RML_MAPPER_PATH(self) -> str:
         return VaultAndEnvConfigResolver().config_resolve()
 
 
+class ELKConfig:
+    @property
+    def ELK_HOST(self) -> str:
+        return VaultAndEnvConfigResolver().config_resolve()
+
+    @property
+    def ELK_PORT(self) -> int:
+        v: str = VaultAndEnvConfigResolver().config_resolve()
+        return int(v) if v is not None else None
+
+    @property
+    def ELK_VERSION(self) -> int:
+        v: str = VaultAndEnvConfigResolver().config_resolve()
+        return int(v) if v is not None else None
+
+
+class LoggingConfig:
+    @property
+    def LOGGING_TYPE(self) -> str:
+        return VaultAndEnvConfigResolver().config_resolve()
+
+
 class XMLProcessorConfig:
 
     @property
     def XML_PROCESSOR_PATH(self) -> str:
         return VaultAndEnvConfigResolver().config_resolve()
 
 
-class TedConfigResolver(MongoDBConfig, RMLMapperConfig, XMLProcessorConfig):
+class TedConfigResolver(MongoDBConfig, RMLMapperConfig, XMLProcessorConfig, ELKConfig, LoggingConfig):
     """
         This class resolve the secrets of the ted-sws project.
     """

ted_sws/core/__init__.py

@@ -3,6 +3,8 @@
 # __init__.py
 # Date:  03/02/2022
 # Author: Eugeniu Costetchi
-# Email: costezki.eugen@gmail.com 
+# Email: costezki.eugen@gmail.com
 
-""" """
+""" 
+Here are instantiated/imported all core/domain services, used throughout the app
+"""

ted_sws/core/adapters/logger.py

@@ -0,0 +1,82 @@
+import abc
+import enum
+import logging
+
+import logstash
+
+from ted_sws import config
+
+
+class LoggingType(enum.Enum):
+    ELK = "ELK"
+    PY = "PY"
+
+
+DOMAIN_LOGGING_TYPES = config.LOGGING_TYPE.split(",") if config.LOGGING_TYPE is not None else [LoggingType.PY.value]
+DOMAIN_LOGGING_TYPE = DOMAIN_LOGGING_TYPES[0]
+
+
+class LoggerABC(abc.ABC):
+    """
+    This abstract class provides methods definitions and infos for available loggers
+    """
+    pass
+
+
+class Logger(LoggerABC):
+    """
+    This class provides common features for available loggers
+    """
+
+    def __init__(self, name: str = __name__):
+        self.name = name
+        self.logger = logging.getLogger(name)
+        self.level = logging.DEBUG
+        self.logger.setLevel(self.level)
+
+    def get_logger(self) -> logging.Logger:
+        return self.logger
+
+    def log(self, msg: str):
+        msg = "\n" + self.name + "\n" + msg + "\n"
+        self.logger.log(self.level, msg)
+
+
+class LoggerFactory:
+    @classmethod
+    def get(cls, logging_type: LoggingType = LoggingType(DOMAIN_LOGGING_TYPE), name: str = __name__):
+        """Factory Method to return the needed Logger, based on logging type/target"""
+        loggers = {
+            LoggingType.ELK: ELKLogger,
+            LoggingType.PY: PYLogger
+        }
+        name = "{logging_type} :: {name}".format(logging_type=logging_type, name=name)
+        return loggers[logging_type](name=name)
+
+
+class ELKLogger(Logger):
+    """
+    LogStash (ELK) Logger
+    """
+
+    def __init__(self, name: str = 'logstash-logger'):
+        super().__init__(name=name)
+
+        host = config.ELK_HOST
+        port = config.ELK_PORT
+        version = config.ELK_VERSION
+
+        self.logger.addHandler(logstash.LogstashHandler(host, port, version=version))
+        # self.logger.addHandler(logstash.TCPLogstashHandler(host, port, version=version))
+
+
+class PYLogger(Logger):
+    """
+    Python Logger
+    """
+
+    def __init__(self, name: str = __name__):
+        super().__init__(name=name)
+
+
+logger = LoggerFactory.get(name="domain-logging")

ted_sws/core/adapters/vault_secrets_store.py

@@ -7,6 +7,7 @@
 
 dotenv.load_dotenv(verbose=True, override=True)
 
+
 class SecretsStoreABC(ABC):
     """
         This class aims to define an interface for obtaining secrets from similar resources as Vault.

ted_sws/core/domain/message_bus.py

@@ -0,0 +1,60 @@
+import logging
+from typing import Callable, List, Dict
+
+from pymessagebus import MessageBus as PyMessageBus
+from pymessagebus.api import Middleware
+from pymessagebus.middleware.logger import get_logger_middleware, LoggingMiddlewareConfig
+
+from ted_sws.core.adapters.logger import logger, Logger
+from ted_sws.core.domain import message_handlers
+from ted_sws.core.model import message
+
+""" 
+Here is instantiated/initialized domain message_bus service
+"""
+
+
+class MessageBus(PyMessageBus):
+    """
+    This class provides additional features to MessageBus
+    """
+    HAS_LOGGING_MIDDLEWARE = True
+    _logger: Logger = logger
+
+    def set_domain_logger(self, _logger: Logger):
+        self._logger = _logger
+
+    def get_domain_logger(self) -> Logger:
+        return self._logger
+
+    def set_middlewares(self, _middlewares: List[Middleware] = None):
+        self._middlewares_chain = self._get_middlewares_callables_chain(
+            _middlewares, self._trigger_handlers_for_message_as_a_middleware
+        )
+
+    def add_handlers(self, handlers: Dict[type, List[Callable]]) -> None:
+        for message_class in handlers:
+            for message_handler in handlers[message_class]:
+                super().add_handler(message_class, message_handler)
+
+
+message_bus = MessageBus()
+
+middlewares: List = []
+if MessageBus.HAS_LOGGING_MIDDLEWARE:
+    logging_middleware_config = LoggingMiddlewareConfig(
+        mgs_received_level=logging.INFO,
+        mgs_succeeded_level=logging.INFO,
+        mgs_failed_level=logging.CRITICAL
+    )
+    logging_middleware = get_logger_middleware(message_bus.get_domain_logger().get_logger(), logging_middleware_config)
+    middlewares.append(logging_middleware)
+
+if len(middlewares) > 0:
+    message_bus.set_middlewares(middlewares)
+
+message_bus.add_handlers({
+    message.Log: [
+        message_handlers.handler_log
+    ]
+})

ted_sws/core/domain/message_handlers.py

@@ -0,0 +1,28 @@
+from ted_sws.core.adapters.logger import LoggerFactory, LoggingType, DOMAIN_LOGGING_TYPES
+from ted_sws.core.model import message
+
+""" 
+Here are defined all core/domain messages' handlers
+"""
+
+
+def handler_log(log: message.Log):
+    """
+    Here is defined the handler for Log Message event
+    :param log:
+    :return:
+    """
+
+    eol = log.format.new_line
+    msg = ""
+    if log.title:
+        msg += ("{title}" + eol).format(title=log.title)
+    if log.messages:
+        msg += ("Messages: " + eol + "{messages}" + eol).format(
+            messages=eol.join(map(lambda m: " - " + m, log.messages))
+        )
+
+    for logging_type_value in DOMAIN_LOGGING_TYPES:
+        _logger = LoggerFactory.get(LoggingType(logging_type_value), name=logging_type_value + "-logging")
+        _logger.log(msg)
+    return msg

ted_sws/core/model/message.py

@@ -0,0 +1,17 @@
+from typing import Optional, List
+
+from pydantic import BaseModel
+
+
+class MessageFormat(BaseModel):
+    new_line: str = "\n"
+
+
+class Message(BaseModel):
+    title: Optional[str] = None
+    messages: Optional[List[str]] = []
+    format: Optional[MessageFormat] = MessageFormat()
+
+
+class Log(Message):
+    pass

ted_sws/core/model/notice.py

@@ -157,6 +157,7 @@ class Notice(WorkExpression):
     _rdf_manifestation: Optional[RDFManifestation] = None
     _mets_manifestation: Optional[METSManifestation] = None
 
+
     @property
     def preprocessed_xml_manifestation(self) -> XMLManifestation:
         return self._preprocessed_xml_manifestation

tests/unit/core/domain/test_message_bus.py

@@ -0,0 +1,39 @@
+#!/usr/bin/python3
+
+"""
+
+"""
+
+from ted_sws.core.domain.message_bus import message_bus
+from ted_sws.core.model.message import Log
+from ted_sws.core.adapters.logger import Logger, LoggingType, LoggerFactory
+
+
+def log_message(logging_type: LoggingType) -> Log:
+    str_longing_type = logging_type.value
+    return Log(
+        title=str_longing_type + " :: test_message_bus_log",
+        messages=[str_longing_type + " :: log_message :: 1", str_longing_type + " :: log_message :: 2"]
+    )
+
+
+def test_message_bus_log(caplog):
+    logging_type = LoggingType.PY
+    log = log_message(logging_type)
+    message_bus.set_domain_logger(LoggerFactory.get(logging_type, name="py-domain"))
+    message_bus.handle(log)
+    if log.title:
+        assert log.title in caplog.text
+    if log.messages:
+        for message in log.messages:
+            assert message in caplog.text
+
+    logging_type = LoggingType.ELK
+    log = log_message(logging_type)
+    message_bus.set_domain_logger(LoggerFactory.get(logging_type, name="elk-domain"))
+    message_bus.handle(log)
+    if log.title:
+        assert log.title in caplog.text
+    if log.messages:
+        for message in log.messages:
+            assert message in caplog.text