无法通过python将数据插入到logstash

如何解决无法通过python将数据插入到logstash

我一直在研究一些日志工具，我找到了 ELK 堆栈，为了快速测试，我使用以下 git 创建测试环境，我正在测试结构化的日志模块如下图：

.
|-- Logger
|   |-- Log
|   |   |-- __init__.py
|   |   |-- metaLogger.py
|   |   `-- singleton.py
|   |-- __main__.py
|   |-- requirements
|   |   `-- requirements.txt
|   `-- __version__.py
|-- MANIFEST
`-- setup.py

使用的包是：

jmespath==0.10.0
python-dateutil==2.8.1
python-logstash==0.4.6
six==1.15.0
urllib3==1.26.3

singleton.py 文件是一个简单的类，只是为了断言我只有一个日志实例：

class Singleton(object):
    """
    Singleton interface:
    https://www.python.org/download/releases/2.2.3/descrintro/#__new__
    """

    def __new__(cls,*args,**kwds):
        it = cls.__dict__.get("__it__")
        if it is not None:
            return it
        cls.__it__ = it = object.__new__(cls)
        it.init(*args,**kwds)
        return it

    def init(self,**kwds):
        pass

metaLogger.py 基本上是大部分配置所在的位置，由于某种原因，由于字符串到字节错误，我不得不从 TCPLogstashHandler 覆盖 makePickle 函数：

import logging
from os import environ,getpid
from logstash import TCPLogstashHandler
from Log.singleton import Singleton
import sys

class ModifiedTCPLogstashHandler(TCPLogstashHandler):
    def makePickle(self,record):
        """Override function to convert string to binary
        """
        return str.encode(self.formatter.format(record)) + b"\n"


class MetaLogger(Singleton):
    """
    Logger metaclass.
    """

    def init(self,robotname: str,loggername: str,loggerlevel):
        self.logger = logging.getLogger(loggername)

        self.logger.setLevel(loggerlevel)
        formatter = logging.Formatter(
            fmt="%(asctime)s | %(levelname)s | %(name)s :: %(message)s",datefmt="%F %H:%M:%S",)

        logstash_handler = ModifiedTCPLogstashHandler(
            "http://localhost",5044,version=1
        )
        logstash_handler.setFormatter(formatter)

        console_handler = logging.StreamHandler()
        console_handler.setFormatter(formatter)

        self.logger.addHandler(console_handler)
        self.logger.addHandler(logstash_handler)
        print(sys.version)
        print(getpid())

    def critical(self,msg: str) -> None:
        self.logger = logging.getLogger(loggername)
        self.logger.critical(msg,exc_info=1)

    def debug(self,msg: str) -> None:
        self.logger = logging.getLogger(loggername)
        self.logger.debug(msg)

    def error(self,msg: str) -> None:
        self.logger = logging.getLogger(loggername)
        self.logger.error(msg,exc_info=1)

    def info(self,msg: str) -> None:
        self.logger = logging.getLogger(loggername)
        self.logger.info(msg)

    def warning(self,msg: str) -> None:
        self.logger = logging.getLogger(loggername)
        self.logger.warning(msg)

最后我有 __main__.py，它基本上是一个实现和测试文件：

from logging import INFO
from Log.metaLogger import MetaLogger


class Logger(object):
    def __init__(
        self,robotname:str = "NULL",loggername: str = "root",loggerlevel=INFO
    ) -> "Logger":
        """
        Logger impelmentation class.
        """
        self.logger = MetaLogger(robotname,loggername,loggerlevel)
        self.loggername = loggername

    def critical(self,msg: str) -> None:
        self.logger.critical(self.loggername,msg)

    def debug(self,msg: str) -> None:
        self.logger.debug(self.loggername,msg)

    def error(self,msg: str) -> None:
        self.logger.error(self.loggername,msg)

    def info(self,msg: str) -> None:
        self.logger.info(self.loggername,msg)

    def warning(self,msg: str) -> None:
        self.logger.warning(self.loggername,msg)


if __name__ == "__main__":
    logger = Logger(loggername="TEST_LOGGER_LOGSTASH")
    logger.info("This is an info")
    logger.warning("this is a warning")
    logger.error("this is an error")
    logger.critical("This is a critical error")

由于某种原因，该模块似乎运行良好，没有一条错误消息，但是当我检查 kibana 仪表板和 elasticsearch 数据库时，我找不到日志，我也从 logstash 中读取了日志，但我可以找不到错误。