标签: logstash-grok

以下是事件消息的示例：

{
"timestamp":"2016-03-29T22:35:44.770750-0400",
"flow_id":45385792,
"in_iface":"eth1",
"event_type":"alert",
"src_ip":"3.3.3.8",
"src_port":21,
"dest_ip":"2.2.2.2",
"dest_port":52934,
"proto":"TCP",
"alert":{
    "action":"allowed",
    "gid":1,
    "signature_id":4027,
    "rev":0,
    "signature":"FTP Successful Login",
    "category":"",
    "severity":3
    },
"payload":"MjU3ICIvaG9tZS9uZXd1c2VyIg0K",
"payload_printable":"257 newuser",
"stream":0,
"packet":"AFBWo0NoAFBWoxZWCABFAABJKDpAAEAGCGcDAwMIAgICAgAVzsbd4MhqOBOjfoAYAOMYcwAAAQEIChHN4EQHnwugMjU3ICIvaG9tZS9uZXd1c2VyIg0K"
}


input 
    beats 
        port => 5044
        codec => json
        type => "SuricataIDPS"

我的 Logstash 配置文件如下：

output 
    elasticsearch 
        hosts => ["localhost:9200"]
        sniffing => true
        manage_template => false
        index => "%{[@metadata][beat]}-%{+YYYY.MM.dd}"
        #document_type => "%{[@metadata][type]}"

我希望能够重命名该字段alert.signature，

我该怎么做？...似乎它不识别该字段...

感谢您的帮助！

埃夫拉特

我有几个安装了 filebeat 的 Web 服务器，并且我希望每个主机有多个索引。

我当前的配置如下

input {
  beats {
    ports => 1337
  }
}

filter {
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}"}
  }
  geoip {
    source => "clientip"
  }
}

output {
  elasticsearch {
  if [beat][hostname] == "luna"
  {
         hosts => "10.0.1.1:9200"
         manage_template => true
         index => "lunaindex-%{+YYYY.MM.dd}"
         document_type => "apache"
  }
  }
}

然而上面的conf结果是

给定的配置无效。原因：第 22 行第 6 列（字节 346）需要 #、=> 之一

这是 if 语句发生的地方。有什么帮助吗？

我希望将以上内容采用嵌套格式

if [beat][hostname] == "lina"
{
index = lina
}
else …

我有一个日志字符串，例如：

2018-08-02 12:02:25.904 [http-nio-8080-exec-1] WARN  o.s.w.s.m.s.DefaultHandlerExceptionResolver.handleTypeMismatch - Failed to bind request element

上面的字符串[http-nio-8080-exec-1]是一个可选字段，它可以出现在一些日志语句中。

我创建了一个 grok 模式，就像网上的一些参考文献一样：

%{TIMESTAMP_ISO8601:timestamp} (\[%{DATA:thread}\])? %{LOGLEVEL:level}%{SPACE}%{JAVACLASS:class}\.%{DATA:method} - %{GREEDYDATA:loggedString}

如果我删除线程名称字符串，它似乎不起作用。

我正在尝试使用 Elastic Stack 解析 Minecraft 日志，但遇到了一个非常奇怪的问题（对我来说可能很奇怪！）

我的日志的所有行都被正确解析，但我_grokparsefailure在每个行中都有标签。

我的logstash管道配置是这样的：

input {
  file {
    path => [ "/path/to/my/log" ]
    #start_position => "beginning"
    tags => ["minecraft"]
  }
}

filter {
  if "minecraft" in [tags] {

#    mutate {
#      gsub => [
#        "message", "\\n", ""
#      ]
#    }



    #############################
    #           Num 1           #
    #############################
    grok {
      match => [ "message", "\[%{TIME:timestamp}] \[(?<originator>[^\/]+)?/%{LOGLEVEL:level}]: %{GREEDYDATA:message}" ]
      overwrite => [ "message" ]
      break_on_match => false
    }


    #############################
    #           Num 2           #
    ############################# …

我试图用grok解析日志文件.我使用的配置允许我解析单个内联事件,但如果是多行的(使用java堆栈跟踪)则不会.

#what i get on KIBANA for a single line:
{
  "_index": "logstash-2015.02.05",
  "_type": "logs",
  "_id": "mluzA57TnCpH-XBRbeg",
  "_score": null,
  "_source": {
    "message": " -  2014-01-14 11:09:35,962 [main] INFO  (api.batch.ThreadPoolWorker)   user.country=US",
    "@version": "1",
    "@timestamp": "2015-02-05T09:38:21.310Z",
    "path": "/root/test2.log",
    "time": "2014-01-14 11:09:35,962",
    "main": "main",
    "loglevel": "INFO",
    "class": "api.batch.ThreadPoolWorker",
    "mydata": "  user.country=US"
  },
  "sort": [
    1423129101310,
    1423129101310
  ]
}

#what i get for a multiline with Stack trace:
  {
  "_index": "logstash-2015.02.05",
  "_type": "logs",
  "_id": "9G6LsSO-aSpsas_jOw",
  "_score": null,
  "_source": {
    "message": "\tat oracle.jdbc.driver.DatabaseError.throwSqlException(DatabaseError.java:20)",
    "@version": "1", …

我使用logstash索引弹性数据库中的一些旧日志文件.我需要kibana/elastic来设置日志文件中的时间戳作为主@timestamp.

我以下列方式使用grok过滤器:%{TIMESTAMP_ISO8601:@timestamp}而elasticsearch将索引的时间设置为主@timestamp,而不是在日志行中写入的时间戳.

知道我在这里做错了什么吗？

谢谢

我们希望使用 Logstash 过滤日志，如果字段不包含“_log”，则删除字段。该remove_field语法可用，但仅适用于删除符合特定条件的字段。

filter {
  grok {
    remove_field => [ "log_" ]
  }
}
# This works for removing the log_ field, we want to remove everything that does NOT match log_.

如果某个字段不符合某个条件，是否也可以删除该字段？

我们尝试使用正则表达式来执行此操作，但这不起作用（是否在某处记录了您不能使用正则表达式？）。删除所有其他字段也是一种选择，但需要付出更多努力。我们希望有人可以帮助我们拟合所有不包含“log_”的字段。

我在SpringBoot应用程序中使用Logback和Logstash。

在logback.xml中，我有一个带有服务名称的属性，如下所示：

<configuration>

<include resource="org/springframework/boot/logging/logback/defaults.xml" />

<include resource="org/springframework/boot/logging/logback/console-appender.xml" />

<property name="spring.application.name" calue="service"/>

<appender name="stash" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
    <destination>localhost:9600</destination>
    <encoder class="net.logstash.logback.encoder.LogstashEncoder"/>
</appender>

<root level="INFO">
    <appender-ref ref="CONSOLE" />
    <appender-ref ref="stash" />
</root>

</configuration>

Logstash conf文件类似于：

input{ tcp{
        port=> 9600
        host=>logstash
    }
}

filter {
grok {
match => {
  "message" =>
  "^%{TIMESTAMP_ISO8601:timestamp}\s+%{LOGLEVEL:level}\s+%{NUMBER:pid}\s+---\s+\[\s*%{USERNAME:thread}\s*\]\s+%{JAVAFILE:class}\s*:\s*%{DATA:themessage}(?:\n+(?<stacktrace>(?:.|\r|\n)+))?$"
}
}
date {
match => [ "timestamp" , "yyyy-MM-dd HH:mm:ss.SSS" ]
}
 mutate {
remove_field => ["@version"]
add_field => {
  "appid" => "%{[path]}"
}
add_field => {
  "levell" => "level"
} …

以下是我的日志

# Time: 2017-11-02T07:41:22.631956Z# User@Host: root[root] @ localhost []  Id:     

我无法为这个词写下grok模式root[root].我想将此值作为单个列.示例:host = root[root].

我想要[我的价值.

这该怎么做 ？

我的日志文件中有多种（三种）类型的日志。其中一种类型有一些自己的打印 + 异常堆栈跟踪。该示例如下所示：

Multiple lines example:
    2018-04-27 10:53:17 [http-nio-8088-exec-4] - ERROR - app-info-exception-info - params:{"cardid":"111111111","txamt":10,"ip":"192.168.16.89","stationcode":"0002","inputuserid":1,"organcode":"99999"} java.lang.NullPointerException: null
        at com.datalook.group.BusinessHandler.handler(BusinessHandler.java:93) ~[classes/:?]
        at com.datalook.group.BusinessGroupController.businessGroup(BusinessGroupController.java:51) [classes/:?]
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:1.8.0_77]
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:1.8.0_77]
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_77]
        at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_77]

我有一个模式来解析它，它是：

#pattern:
(?<timestamp>[\d\-\s\:]+)\s\[(?<threadname>[\w\-\d]+)\]\s-\s(?<loglevel>[\w]+)\s\-\s(?<appinfo>app-info-exception-info)\s-\s(?<params>params):(?<jsonstr>[\"\w\d\,\:\.\{\}]+)\s(?<exceptionname>[\w\d\.]+Exception):\s(?<exceptiondetail>[\w\d\.]+)\n\t(?<extralines>at[\s\w\.\d\~\?\n\t\(\)\_\[\]\/\:\-]+)\n

模式在解析多行异常堆栈跟踪时有错误（实际上不是错误，但没有完全解析或按预期解析），主要在最后两部分（exceptiondetail（在这种情况下为 null）和 extralines（那些以空格或制表符开头的行加上 'at '，或第一行堆栈跟踪之后的行））。有比我更好的主意吗？

在 filebeat.yml 中，我配置了以下内容：

# The regexp Pattern that has to be matched. The example pattern matches all lines starting with [
  multiline.pattern: '^[[:space:]]'

  # Defines if the pattern set under pattern should be negated or not. …

        第一| 2nd|3rd |4th |5th |6th |7th |8th |2012.07.12 05:31:04 |10th |ProductDir: C:\samplefiles\test\storage\4.0 (LF)
C:\samplefiles\test\storage\5.0 (LF)
样本目录：(LF)

注意：LF -> 换行被附加

我已经尝试过以下选项..似乎没有任何效果

1. match => [ "message", "(?m)....
2. (?<message>(.|\r|\n)*)
3. Greedydata 也无法工作，因为它不考虑新线。
4. mutate {gsub => ["message", "\n", "LINE_BREAK"] }

5. 编解码器 => 多行 { 模式 => "^\s" 否定 => true 什么 => 上一个 }

我将Web和API日志合并在一起，我想将其单独保存在elasticsearch中。所以我想编写一种模式，如果请求是 API，那么如果过去应该执行，请求是 Web，那么应该执行日志的一部分。

以下是一些 Web 和 API 日志。

00:06:27,778 INFO  [stdout] (ajp--0.0.0.0-8009-38) 00:06:27.777 [ajp--0.0.0.0-8009-38] INFO  c.r.s.web.rest.WidgetController - Method getWidgetDetails() started to get widget details.
00:06:27,783 INFO  [stdout] (ajp--0.0.0.0-8009-38) ---> HTTP GET http://api.survey.me/v1/getwidgetdetails?profileName=jeremy-steffens&profileLevel=INDIVIDUAL&companyProfileName=premier-nationwide-lending&hideHistory=true
00:06:27,817 INFO  [stdout] (ajp--0.0.0.0-8009-38) <--- HTTP 200 http://api.survey.me/v1/getwidgetdetails?profileName=jeremy-steffens&profileLevel=INDIVIDUAL&companyProfileName=premier-nationwide-lending&hideHistory=true (29ms)
00:06:27,822 INFO  [stdout] (ajp--0.0.0.0-8009-38) 00:06:27.822 [ajp--0.0.0.0-8009-38] INFO  c.r.s.web.rest.WidgetController - Method getWidgetDetails() finished.
00:06:27,899 INFO  [stdout] (ajp--0.0.0.0-8009-40) 00:06:27.899 [ajp--0.0.0.0-8009-40] INFO  c.r.s.web.controller.LoginController - Inside initLoginPage() of LoginController

我试图写条件但它不起作用。它仅适用于线程名称。在线程之后，我有多种类型的日志，因此无法在没有 if 条件的情况下写入。

(?:%{TIME:CREATED_ON})(?:%{SPACE})%{WORD:LEVEL}%{SPACE}\[%{NOTSPACE}\]%{SPACE}\(%{NOTSPACE:THREAD}\)

有人可以给我建议吗？