以纯文本格式从 OUD 日志中提取信息的最佳方法

Question

我想分析来自 OUD（Oracle Unified Directory）的日志并使用 Kibana 创建一些图表。我在 OUD 服务器上安装了一个 filebeat 代理，该代理接收日志然后将它们发送到 elasticsearch。没有任何预配置的仪表板或模块，因此我需要从头开始创建所有内容。我的问题是日志采用纯文本格式，全部存储在一个字段“消息”中。它们可以有 5 种类型（BIND、CONNECT、SEARCH、DISCONNECT、UNBIND）。以下是一些例子：

[2024-06-19T16:00:00.479+02:00] [OUD] [TRACE] [OUD-24641548] [PROTOCOL] [host: XXXX.en] [nwaddr: X.X.X.X] [tid: 281] [userId: ouduser] [ecid: XXX,0:1] [category: REQ] [conn: 1694606] [op: 0] [msgID: 1] [bindType: SASL] [bindMech: DIGEST-MD5] [dn: ] BIND

[2024-06-19T16:00:00.489+02:00] [OUD] [TRACE] [OUD-24641551] [PROTOCOL] [host: XXXX.en] [nwaddr: X.X.X.X] [tid: 289] [userId: ouduser] [ecid: XXX,0:1] [category: REQ] [conn: -1] [op: 3105334] [msgID: 3105335] [base: XXX] [scope: base] [filter: (objectClass=*)] [attrs: ds-privilege-name,+,*] SEARCH

[2024-06-19T15:59:00.467+02:00] [OUD] [TRACE] [OUD-24641551] [PROTOCOL] [host: XXXX.en] [nwaddr: X.X.X.X] [tid: 281] [userId: ouduser] [ecid: XXX,0:1] [category: REQ] [conn: 1694593] [op: 4] [msgID: 5] [base: x] [scope: sub] [filter: (sAMAccountName=USER)] [attrs: dn,authPassword,orclPassword,orclguid] SEARCH

[2024-06-19T15:59:00.438+02:00] [OUD] [TRACE] [OUD-24641547] [PROTOCOL] [host: XXXX.en] [nwaddr: X.X.X.X] [tid: 105] [userId: ouduser] [ecid: XXX,0] [conn: 1694595] [reason: Client Disconnect] DISCONNECT

我尝试使用 Grok 来提取字段，但这是一场噩梦，因为找到模式是不可能的。基本上我想要每个日志都是这样的：

{
 type : BIND,
 timestamp : 2024-06-19T16:00:00.479+02:00,
 userId: ouduser,
 host: XXXX.en,
 //I want to be able to choose which field I include
 etc...
}

• 有没有一种简单的方法可以使用 Grok 来做到这一点？
• 如果没有，是否有一个我可以使用的工具，这样我就不需要给出日志所需的确切模式（因为在我的情况下，即使在相同类型的日志之间也是不可能的）

感谢您的帮助！

Answer 1

Tldr；

Grok 或 Dissect 将是您的首选工具。

由于elastic似乎没有集成OUD，我相信你必须自己解析它。

锅炉板

请在下面找到我使用的一些样板代码：

• 您的意见
• Kibana > 开发工具 > Grok 调试器

 input {
      generator {
        lines => [
          "[2024-06-19T16:00:00.479+02:00] [OUD] [TRACE] [OUD-24641548] [PROTOCOL] [host: XXXX.en] [nwaddr: X.X.X.X] [tid: 281] [userId: ouduser] [ecid: XXX,0:1] [category: REQ] [conn: 1694606] [op: 0] [msgID: 1] [bindType: SASL] [bindMech: DIGEST-MD5] [dn: ] BIND",
          "[2024-06-19T16:00:00.489+02:00] [OUD] [TRACE] [OUD-24641551] [PROTOCOL] [host: XXXX.en] [nwaddr: X.X.X.X] [tid: 289] [userId: ouduser] [ecid: XXX,0:1] [category: REQ] [conn: -1] [op: 3105334] [msgID: 3105335] [base: XXX] [scope: base] [filter: (objectClass=*)] [attrs: ds-privilege-name,+,*] SEARCH",
          "[2024-06-19T15:59:00.467+02:00] [OUD] [TRACE] [OUD-24641551] [PROTOCOL] [host: XXXX.en] [nwaddr: X.X.X.X] [tid: 281] [userId: ouduser] [ecid: XXX,0:1] [category: REQ] [conn: 1694593] [op: 4] [msgID: 5] [base: x] [scope: sub] [filter: (sAMAccountName=USER)] [attrs: dn,authPassword,orclPassword,orclguid] SEARCH",
          "[2024-06-19T15:59:00.438+02:00] [OUD] [TRACE] [OUD-24641547] [PROTOCOL] [host: XXXX.en] [nwaddr: X.X.X.X] [tid: 105] [userId: ouduser] [ecid: XXX,0] [conn: 1694595] [reason: Client Disconnect] DISCONNECT"
        ]
        count => 1
      }
    }

filter { 
  grok {
    match => {
      "message" => [
        "\[%{DATA:@timestamp}\] \[%{WORD:service.name}\] \[%{WORD:log.level}\] \[%{DATA:service.id}\] \[%{DATA:unknw}\] \[host:%{DATA:host.domain}\] \[nwaddr:%{DATA:host.ip}\] \[tid:%{DATA:process.thread.id}\] \[userId:%{DATA:user.id}\] \[ecid:%{DATA:transaction.id}\] \[category:%{DATA:event.category}\] \[conn:%{DATA:network.connection.id}\] \[op:%{DATA:operation.id}\] \[msgID:%{DATA:event.id}\] \[bindType:%{DATA:auth.type}\] \[bindMech:%{DATA:auth.mech}\] \[dn:%{DATA:dn}\] %{WORD:event.action}",
        "\[%{DATA:@timestamp}\] \[%{WORD:service.name}\] \[%{WORD:log.level}\] \[%{DATA:service.id}\] \[%{DATA:unknw}\] \[host:%{DATA:host.domain}\] \[nwaddr:%{DATA:host.ip}\] \[tid:%{DATA:process.thread.id}\] \[userId:%{DATA:user.id}\] \[ecid:%{DATA:transaction.id}\] \[category:%{DATA:event.category}\] \[conn:%{DATA:network.connection.id}\] \[op:%{DATA:operation.id}\] \[msgID:%{DATA:event.id}\] \[base:%{DATA:base}\] \[scope:%{DATA:scope}\] \[filter:%{DATA:filter}\] \[attrs:%{DATA:attrs}\] %{WORD:event.action}",
        "\[%{DATA:@timestamp}\] \[%{WORD:service.name}\] \[%{WORD:log.level}\] \[%{DATA:service.id}\] \[%{DATA:unknw}\] \[host:%{DATA:host.domain}\] \[nwaddr:%{DATA:host.ip}\] \[tid:%{DATA:process.thread.id}\] \[userId:%{DATA:user.id}\] \[ecid:%{DATA:transaction.id}\] \[conn:%{DATA:network.connection.id}\] \[reason:%{DATA:reason}\] %{WORD:event.action}"
      ]
    }
  }
}

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