elasticsearch - 在将完整日志发送到logstash/elasticsearch之前，将一半日志发送到RabbitMQ的方法

Question

我有几个功能，每个功能都创建特定于一个事务的日志。这是一个多线程应用程序，因此func1的函数条目可以对进行的事务是随机的，但对于单个事务，它将仅通过func1，func2和func3顺序进行。

func1(transactionId) {
     log("%d Now in func1", transactionId);
}

func2(transactionId) {
     log("%d Now in func2", transactionId);
}

func3(transactionId) {
     log("%d Now in func3", transactionId);
}

现在，我想一次只为每个事务一次写入logstash；那是

 1 Now in func1 Now in func2 Now in fun3

然后最终需要进行 flex 搜索；

我当时正在考虑将一半的事务日志写入RabbitMQ临时队列，然后在完成事务后，将其提交给RabbitMQ生产者队列，以将消息发送至logstash。

喜欢

func1(transactionId) {
     add2RMQ(transactionId, "Now in func1");
}

func2(transactionId) {
     add2RMQ("transactionId, "Now in func2");
}

func3(transactionId) {
      add2RMQ("transactionId, "Now in func3");
      /* Last point of transaction */
      commit2RMQ(transactionId);
}

当commit2RMQ执行logstash时，应该会收到特定于事务的完整消息，以写入Elasticsearch。

题:

解决此问题以立即将特定于事务的数据发送到elasticsearch的正确解决方案是什么？

我们可以用RabbitMQ解决这个问题吗？如果是这样，我需要为此使用什么正确的API？

没有RabbitMQ，但只有Logstash和Elasticsearch，有什么方法可以实现相同目的吗？

我不想使用elasticsearch更新API，因为它可能会为每个特定于事务处理的日志消息消耗大量搜索操作。

Answer 1

尝试汇总与单个事务相关的不同日志行不是一个简单的问题，尤其是如果您将消息队列系统添加到要作为汇总日志的中间存储库的消息队列系统中。我将采用一种不涉及其他子系统(例如RabbitMQ)的不同方式。

此外，如果尝试将多条日志行连接到一个日志行中，则会丢失每条日志行可以提供的详细信息，例如每个函数执行所花费的时间。如果func2和func3分别抛出异常，还会发生什么？您是否应该存储仅由func1或func1和func2组成的部分日志？

我要写的内容可能可以转换为任何语言和任何日志记录解决方案，但是出于说明的目的，我假设您的程序是用Java编写的，并且您正在使用Log4J。

因此，我将利用Log4J's Mapped Diagnostic Context(MDC)来在每个日志行中存储您的事务ID(以及潜在的其他数据，例如用户名等)。这样，您可以轻松检索与单个事务有关的所有日志行。这样做的好处是您不必聚合任何东西，您只需提供足够的上下文信息，以便Kibana稍后可以为您完成。

在伪代码中，您是将交易ID直接添加到消息中。为此使用MDC而不是将ID登录到消息中的优点是，它使您不必解析Logstash中的所有消息来重新发现创建日志行时已经知道的事务ID。

因此，我们的想法是，在您的代码中，一旦有了事务ID，便将其添加到当前的每线程日志上下文中，如下所示:

import org.apache.log4j.MDC;

...
func1(transactionId) {
     // add the transaction ID to the logging context
     MDC.put("transactionID", transactionId);
     log("Now in func1");
}

func2(transactionId) {
     log("Now in func2");
}

func3(transactionId) {
     log("Now in func3");
}

然后，在Log4J配置文件中，可以使用 %X{transactionID}模式指定附加程序以便存储它，在这种情况下，我将其添加到线程名称之后，但是可以将其放置在任意位置:

log4j.appender.consoleAppender.layout.ConversionPattern = %d [%t] [%X{transactionID}] %5p %c - %m%n

您的日志将类似于以下内容:

2015-09-28T05:07:28.425Z [http-8084-2] [625562271762]  INFO YourClass - Now in func1
2015-09-28T05:07:29.776Z [http-8084-2] [625562271762]  INFO YourClass - Now in func2
2015-09-28T05:07:30.652Z [http-8084-2] [625562271762]  INFO YourClass - Now in func3
                                              ^
                                              |
                                  the transaction ID is here

当您有这样的日志行时，通过Logstash grok过滤器检索事务ID并将其存储在自己的logstash索引中的自己的 transactionID字段中就很容易了。然后，在Kibana中，您可以搜索事务ID并按时间戳记desc进行排序，然后将显示该事务的所有上下文。

试一试!