elasticsearch - 使用日期过滤器将 Logstash 日期解析为时间戳

Question

好吧，在四处寻找之后，我找不到解决我的问题的方法，因为它“应该”有效，但显然没有。我在 Ubuntu 14.04 LTS 机器上使用 Logstash 1.4.2-1-2-2c0f5a1，我收到如下消息:

2014-08-05 10:21:13,618 [17] INFO  Class.Type - This is a log message from the class:
  BTW, I am also multiline

在输入配置中，我确实有一个 multiline 编解码器并且事件被正确解析。我还将事件文本分成几个部分，以便于阅读。

最后，如在 Kibana 中所见，我获得了如下内容(JSON View ):

{
  "_index": "logstash-2014.08.06",
  "_type": "customType",
  "_id": "PRtj-EiUTZK3HWAm5RiMwA",
  "_score": null,
  "_source": {
    "@timestamp": "2014-08-06T08:51:21.160Z",
    "@version": "1",
    "tags": [
      "multiline"
    ],
    "type": "utg-su",
    "host": "ubuntu-14",
    "path": "/mnt/folder/thisIsTheLogFile.log",
    "logTimestamp": "2014-08-05;10:21:13.618",
    "logThreadId": "17",
    "logLevel": "INFO",
    "logMessage": "Class.Type - This is a log message from the class:\r\n  BTW, I am also multiline\r"
  },
  "sort": [
    "21",
    1407315081160
  ]
}

你可能已经注意到我放了一个“;”在时间戳中。原因是我希望能够使用时间戳字符串对日志进行排序，显然 logstash 并不擅长(例如:http://www.elasticsearch.org/guide/en/elasticsearch/guide/current/multi-fields.html)。

我尝试以多种方式使用 date 过滤器，但均未成功，但显然没有成功。

date {
            locale => "en"
            match => ["logTimestamp", "YYYY-MM-dd;HH:mm:ss.SSS", "ISO8601"]
            timezone => "Europe/Vienna"
            target => "@timestamp"
            add_field => { "debug" => "timestampMatched"}
        }

因为我读到如果字符串不严格符合 ISO 8601 标准，Joda 库可能会出现问题(非常挑剔，需要一个 T，参见 https://logstash.jira.com/browse/LOGSTASH-180)，我也尝试使用 mutate 将字符串转换为类似 2014-08-05T10:21:13.618 的内容，然后使用 "YYYY-MM-dd'T'HH:mm:ss .SSS”。那也没有用。

我不想手动将时间设置为 +02:00，因为那样会导致夏令时出现问题。

在任何这些情况下，事件都会转到 elasticsearch，但 date 显然什么都不做，因为 @timestamp 和 logTimestamp 是不同的并且没有添加 debug 字段。

知道如何使 logTime 字符串正确排序吗？我专注于将它们转换为适当的时间戳，但也欢迎任何其他解决方案。

如下图所示:

当对 @timestamp 进行排序时，elasticsearch 可以正确地完成它，但由于这不是“真正的”日志时间戳，而是读取 logstash 事件时，我需要(显然)是还可以对 logTimestamp 进行排序。这就是随后的输出。显然没那么有用:

欢迎任何帮助!如果我忘记了一些可能有用的信息，请告诉我。

更新:

这是最终起作用的过滤器配置文件:

# Filters messages like this:
# 2014-08-05 10:21:13,618 [17] INFO  Class.Type - This is a log message from the class:
#  BTW, I am also multiline

# Take only type- events (type-componentA, type-componentB, etc)
filter {
    # You cannot write an "if" outside of the filter!
    if "type-" in [type] {
        grok {
            # Parse timestamp data. We need the "(?m)" so that grok (Oniguruma internally) correctly parses multi-line events
            patterns_dir => "./patterns"
            match => [ "message", "(?m)%{TIMESTAMP_ISO8601:logTimestampString}[ ;]\[%{DATA:logThreadId}\][ ;]%{LOGLEVEL:logLevel}[ ;]*%{GREEDYDATA:logMessage}" ]
        }

        # The timestamp may have commas instead of dots. Convert so as to store everything in the same way
        mutate {
            gsub => [
                # replace all commas with dots
                "logTimestampString", ",", "."
                ]
        }

        mutate {
            gsub => [
                # make the logTimestamp sortable. With a space, it is not! This does not work that well, in the end
                # but somehow apparently makes things easier for the date filter
                "logTimestampString", " ", ";"
                ]
        }

        date {
            locale => "en"
            match => ["logTimestampString", "YYYY-MM-dd;HH:mm:ss.SSS"]
            timezone => "Europe/Vienna"
            target => "logTimestamp"
        }
    }
}

filter {
    if "type-" in [type] {
        # Remove already-parsed data
        mutate {
            remove_field => [ "message" ]
        }
    }
}

Answer 1

我已经测试了您的date 过滤器。它对我有用!

这是我的配置

input {
    stdin{}
}

filter {
    date {
        locale => "en"
        match => ["message", "YYYY-MM-dd;HH:mm:ss.SSS"]
        timezone => "Europe/Vienna"
        target => "@timestamp"
        add_field => { "debug" => "timestampMatched"}
   }
}

output {
    stdout {
            codec => "rubydebug"
    }
}

我使用这个输入:

2014-08-01;11:00:22.123

输出是:

{
   "message" => "2014-08-01;11:00:22.123",
  "@version" => "1",
"@timestamp" => "2014-08-01T09:00:22.123Z",
      "host" => "ABCDE",
     "debug" => "timestampMatched"
}

因此，请确保您的 logTimestamp 具有正确的值。这可能是其他问题。或者您能否提供您的日志事件和 logstash 配置以供更多讨论。谢谢。