GrokReader

GrokReader Controller Service 提供了一种解析和结构化无结构文本输入的手段， 例如日志文件。Grok 允许用户为正则表达式添加命名结构，以便组合出更易管理 和使用的表达式。该 Controller Service 包含一个必需属性和若干可选属性。 名为 Grok Pattern File 的属性指定一个包含可用于解析日志数据的 Grok Patterns 的文件。未指定时使用默认的 patterns 文件，其内容见下文。 还有用于指定解析数据所用 schema 的属性。Schema 不是必需的；不过如果不 指定，解析出的记录会包含 Grok 表达式中的所有字段（见下文说明），且所有 字段都是 String 类型。如果选择了 schema，字段可以声明为其他兼容类型 （如数值）。此外，如果 schema 不包含解析数据中的某个字段，该字段会被 忽略。这可以用来过滤掉不感兴趣的字段。

注意：还会添加一个 _raw 字段，用于保存原始消息。

必需属性名为 Grok Expression，指定如何解析每条传入记录。做法是提供 一个 Grok 表达式，例如： %{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} \[%{DATA:thread}\] %{DATA:class} %{GREEDYDATA:message}。 这个表达式可以解析 ACAL Lynxus 日志消息。原理是指定每行以 TIMESTAMP_ISO8601 模式开头（这是在默认 Grok Patterns 文件中定义的 正则表达式），匹配该模式的值被命名为 timestamp。因此，匹配该模式的值 会被赋给该 Controller Service 生成的记录中名为 timestamp 的字段。

无匹配行为（No Match Behavior）

如果 FlowFile 中遇到不匹配所配置 Grok 表达式的行，则认为该行是上一条 消息的一部分。如果该行是堆栈跟踪的开始，则整个堆栈跟踪会被读取并赋给 名为 STACK_TRACE 的字段。否则，该行按 "No Match Behavior" 属性定义 的值处理。

Append to Previous Message（追加到上一条消息）

该行被追加到 Grok 表达式中定义的最后一个字段。通常这样做是因为最后一个 字段是 'message' 类型的字段，可以包含换行。

Skip Line（跳过该行）

该行被完全丢弃。

Raw Line（原始行）

除 _raw 字段外所有字段都为 null，_raw 字段包含该行，便于下游进一步处理。

示例

假设有两条消息（<6>Feb 28 12:00:00 192.168.0.1 aliyun[11111]: [error] Syslog test 和 This is a bad message...），Grok 表达式为 <%{POSINT:priority}>%{SYSLOGTIMESTAMP:timestamp} %{SYSLOGHOST:hostname} %{WORD:ident}%{GREEDYDATA:message}； 假设使用 JSON Writer，会生成以下结果：

Append to Previous Message

[
  {
    "priority": "6",
    "timestamp": "Feb 28 12:00:00",
    "hostname": "192.168.0.1",
    "ident": "aliyun",
    "message": "[11111]: [error] Syslog test\nThis is a bad message...",
    "stackTrace": null,
    "_raw": "<6>Feb 28 12:00:00 192.168.0.1 aliyun[11111]: [error] Syslog test\nThis is a bad message..."
  }
]

Skip Line

[
  {
    "priority": "6",
    "timestamp": "Feb 28 12:00:00",
    "hostname": "192.168.0.1",
    "ident": "aliyun",
    "message": "[11111]: [error] Syslog test",
    "stackTrace": null,
    "_raw": "<6>Feb 28 12:00:00 192.168.0.1 aliyun[11111]: [error] Syslog test"
  }
]

Raw Line

[
  {
    "priority": "6",
    "timestamp": "Feb 28 12:00:00",
    "hostname": "192.168.0.1",
    "ident": "aliyun",
    "message": "[11111]: [error] Syslog test",
    "stackTrace": null,
    "_raw": "<6>Feb 28 12:00:00 192.168.0.1 aliyun[11111]: [error] Syslog test"
  },
  {
    "priority": null,
    "timestamp": null,
    "hostname": null,
    "ident": null,
    "message": null,
    "stackTrace": null,
    "_raw": "This is a bad message..."
  }
]

Schema 与类型强制转换

从传入数据解析记录时，数据会被拆分为各个字段。每个字段随后会按字段名与 配置的 schema 进行比对，以确定数据的类型。如果字段不在 schema 中，该字段 会从记录中省略。如果在 schema 中找到了该字段，接收到的数据的数据类型会 与 schema 指定的数据类型比较。如果类型一致，字段值原样使用。如果 schema 表明该字段应为其他类型，Controller Service 会尝试把数据强制转换为 schema 指定的类型。如果无法强制转换，会抛出异常。

尝试把一个字段值从一种数据类型强制转换为另一种时，适用以下规则：

• 任何数据类型都可以强制转换为 String 类型。
• 任何数值类型（Byte、Short、Int、Long、Float、Double）都可以强制转换为其他数值类型。
• 任何数值都可以强制转换为 Date、Time 或 Timestamp 类型，方式是把 Long 值 视为自纪元（1970 年 1 月 1 日午夜，GMT）以来的毫秒数。
• 如果 String 值的格式与配置的 "Date Format"、"Time Format" 或 "Timestamp Format" 匹配，就可以强制转换为 Date、Time 或 Timestamp 类型。
• 如果值的类型合适，String 值可以强制转换为数值。例如，String 值 8 可以强制转换为任何数值类型；但 String 值 8.2 只能强制转换为 Double 或 Float 类型，不能转为 Integer。
• String 值 "true" 或 "false"（不区分大小写）可以强制转换为 Boolean 值。
• 非空的 String 值可以强制转换为 Char 类型。如果字符串包含多个字符， 只使用第一个字符，其余忽略。
• 任何"日期/时间"类型（Date、Time、Timestamp）都可以强制转换为其他"日期/时间"类型。
• 任何"日期/时间"类型都可以强制转换为 Long 类型，表示自纪元（1970 年 1 月 1 日午夜，GMT）以来的毫秒数。
• 任何"日期/时间"类型都可以强制转换为 String。String 的格式就是对应属性 （Date Format、Time Format、Timestamp Format）所配置的 DateFormat。

如果尝试在两种数据类型之间强制转换时上述规则都不适用，转换会失败并抛出异常。

示例

举例来说，假设该 Controller Service 配置了以下属性：

再假设有一个内容如下的带注释的 FlowFile：

2016-08-04 13:26:32,473 INFO [Leader Election Notification Thread-1] o.a.n.c.l.e.CuratorLeaderElectionManager cn.org.acal.lynxus.controller.leader.election.CuratorLeaderElectionManager$ElectionListener@1fa27ea5 has been interrupted; no longer leader for role 'Cluster Coordinator'
2016-08-04 13:26:32,474 ERROR [Leader Election Notification Thread-2] o.apache.lynxus.controller.FlowController One
Two
Three
cn.org.acal.lynxus.exception.UnitTestException: Testing to ensure we are able to capture stack traces
        at cn.org.acal.lynxus.cluster.coordination.node.NodeClusterCoordinator.getElectedActiveCoordinatorAddress(NodeClusterCoordinator.java:185)
        at cn.org.acal.lynxus.cluster.coordination.node.NodeClusterCoordinator.getElectedActiveCoordinatorAddress(NodeClusterCoordinator.java:185)
        at cn.org.acal.lynxus.cluster.coordination.node.NodeClusterCoordinator.getElectedActiveCoordinatorAddress(NodeClusterCoordinator.java:185)
        at cn.org.acal.lynxus.cluster.coordination.node.NodeClusterCoordinator.getElectedActiveCoordinatorAddress(NodeClusterCoordinator.java:185)
        at cn.org.acal.lynxus.cluster.coordination.node.NodeClusterCoordinator.getElectedActiveCoordinatorAddress(NodeClusterCoordinator.java:185)
        at cn.org.acal.lynxus.cluster.coordination.node.NodeClusterCoordinator.getElectedActiveCoordinatorAddress(NodeClusterCoordinator.java:185)
        at cn.org.acal.lynxus.cluster.coordination.node.NodeClusterCoordinator.getElectedActiveCoordinatorAddress(NodeClusterCoordinator.java:185)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_45]
	at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308) [na:1.8.0_45]
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180) [na:1.8.0_45]
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294) [na:1.8.0_45]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [na:1.8.0_45]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [na:1.8.0_45]
        at java.lang.Thread.run(Thread.java:745) [na:1.8.0_45]
Caused by: cn.org.acal.lynxus.exception.UnitTestException: Testing to ensure we are able to capture stack traces
    at cn.org.acal.lynxus.cluster.coordination.node.NodeClusterCoordinator.getElectedActiveCoordinatorAddress(NodeClusterCoordinator.java:185)
    at cn.org.acal.lynxus.cluster.coordination.node.NodeClusterCoordinator.getElectedActiveCoordinatorAddress(NodeClusterCoordinator.java:185)
    at cn.org.acal.lynxus.cluster.coordination.node.NodeClusterCoordinator.getElectedActiveCoordinatorAddress(NodeClusterCoordinator.java:185)
    at cn.org.acal.lynxus.cluster.coordination.node.NodeClusterCoordinator.getElectedActiveCoordinatorAddress(NodeClusterCoordinator.java:185)
    at cn.org.acal.lynxus.cluster.coordination.node.NodeClusterCoordinator.getElectedActiveCoordinatorAddress(NodeClusterCoordinator.java:185)
    ... 12 common frames omitted
2016-08-04 13:26:35,475 WARN [Curator-Framework-0] org.apache.curator.ConnectionState Connection attempt unsuccessful after 3008 (greater than max timeout of 3000). Resetting connection and trying again with a new connection.

此时，该 FlowFile 会生成 3 条不同的记录。第一条记录包含以下值：

第二条记录包含以下值：

第三条记录包含以下值：

默认 Patterns

默认 Grok Pattern 文件中包含以下 patterns：

# Log Levels
LOGLEVEL ([Aa]lert|ALERT|[Tt]race|TRACE|[Dd]ebug|DEBUG|[Nn]otice|NOTICE|[Ii]nfo|INFO|[Ww]arn?(?:ing)?|WARN?(?:ING)?|[Ee]rr?(?:or)?|ERR?(?:OR)?|[Cc]rit?(?:ical)?|CRIT?(?:ICAL)?|[Ff]atal|FATAL|[Ss]evere|SEVERE|EMERG(?:ENCY)?|[Ee]merg(?:ency)?)|FINE|FINER|FINEST|CONFIG

# Syslog Dates: Month Day HH:MM:SS
SYSLOGTIMESTAMP %{MONTH} +%{MONTHDAY} %{TIME}
PROG (?:[\w._/%-]+)
SYSLOGPROG %{PROG:program}(?:\[%{POSINT:pid}\])?
SYSLOGHOST %{IPORHOST}
SYSLOGFACILITY <%{NONNEGINT:facility}.%{NONNEGINT:priority}>
HTTPDATE %{MONTHDAY}/%{MONTH}/%{YEAR}:%{TIME} %{INT}

# Months: January, Feb, 3, 03, 12, December
MONTH \b(?:Jan(?:uary)?|Feb(?:ruary)?|Mar(?:ch)?|Apr(?:il)?|May|Jun(?:e)?|Jul(?:y)?|Aug(?:ust)?|Sep(?:tember)?|Oct(?:ober)?|Nov(?:ember)?|Dec(?:ember)?)\b
MONTHNUM (?:0?[1-9]|1[0-2])
MONTHNUM2 (?:0[1-9]|1[0-2])
MONTHDAY (?:(?:0[1-9])|(?:[12][0-9])|(?:3[01])|[1-9])

# Days: Monday, Tue, Thu, etc...
DAY (?:Mon(?:day)?|Tue(?:sday)?|Wed(?:nesday)?|Thu(?:rsday)?|Fri(?:day)?|Sat(?:urday)?|Sun(?:day)?)

# Years?
YEAR (?>\d\d){1,2}
HOUR (?:2[0123]|[01]?[0-9])
MINUTE (?:[0-5][0-9])
# '60' is a leap second in most time standards and thus is valid.
SECOND (?:(?:[0-5]?[0-9]|60)(?:[:.,][0-9]+)?)
TIME (?!<[0-9])%{HOUR}:%{MINUTE}(?::%{SECOND})(?![0-9])

# datestamp is YYYY/MM/DD-HH:MM:SS.UUUU (or something like it)
DATE_US_MONTH_DAY_YEAR %{MONTHNUM}[/-]%{MONTHDAY}[/-]%{YEAR}
DATE_US_YEAR_MONTH_DAY %{YEAR}[/-]%{MONTHNUM}[/-]%{MONTHDAY}
DATE_US %{DATE_US_MONTH_DAY_YEAR}|%{DATE_US_YEAR_MONTH_DAY}
DATE_EU %{MONTHDAY}[./-]%{MONTHNUM}[./-]%{YEAR}
ISO8601_TIMEZONE (?:Z|[+-]%{HOUR}(?::?%{MINUTE}))
ISO8601_SECOND (?:%{SECOND}|60)
TIMESTAMP_ISO8601 %{YEAR}-%{MONTHNUM}-%{MONTHDAY}[T ]%{HOUR}:?%{MINUTE}(?::?%{SECOND})?%{ISO8601_TIMEZONE}?
DATE %{DATE_US}|%{DATE_EU}
DATESTAMP %{DATE}[- ]%{TIME}
TZ (?:[PMCE][SD]T|UTC)
DATESTAMP_RFC822 %{DAY} %{MONTH} %{MONTHDAY} %{YEAR} %{TIME} %{TZ}
DATESTAMP_RFC2822 %{DAY}, %{MONTHDAY} %{MONTH} %{YEAR} %{TIME} %{ISO8601_TIMEZONE}
DATESTAMP_OTHER %{DAY} %{MONTH} %{MONTHDAY} %{TIME} %{TZ} %{YEAR}
DATESTAMP_EVENTLOG %{YEAR}%{MONTHNUM2}%{MONTHDAY}%{HOUR}%{MINUTE}%{SECOND}


POSINT \b(?:[1-9][0-9]*)\b
NONNEGINT \b(?:[0-9]+)\b
WORD \b\w+\b
NOTSPACE \S+
SPACE \s*
DATA .*?
GREEDYDATA .*
QUOTEDSTRING (?>(?<!\\)(?>"(?>\\.|[^\\"]+)+"|""|(?>'(?>\\.|[^\\']+)+')|''|(?>`(?>\\.|[^\\`]+)+`)|``))
UUID [A-Fa-f0-9]{8}-(?:[A-Fa-f0-9]{4}-){3}[A-Fa-f0-9]{12}

USERNAME [a-zA-Z0-9._-]+
USER %{USERNAME}
INT (?:[+-]?(?:[0-9]+))
BASE10NUM (?<![0-9.+-])(?>[+-]?(?:(?:[0-9]+(?:\.[0-9]+)?)|(?:\.[0-9]+)))
NUMBER (?:%{BASE10NUM})
BASE16NUM (?<![0-9A-Fa-f])(?:[+-]?(?:0x)?(?:[0-9A-Fa-f]+))
BASE16FLOAT \b(?<![0-9A-Fa-f.])(?:[+-]?(?:0x)?(?:(?:[0-9A-Fa-f]+(?:\.[0-9A-Fa-f]*)?)|(?:\.[0-9A-Fa-f]+)))\b

# Networking
MAC (?:%{CISCOMAC}|%{WINDOWSMAC}|%{COMMONMAC})
CISCOMAC (?:(?:[A-Fa-f0-9]{4}\.){2}[A-Fa-f0-9]{4})
WINDOWSMAC (?:(?:[A-Fa-f0-9]{2}-){5}[A-Fa-f0-9]{2})
COMMONMAC (?:(?:[A-Fa-f0-9]{2}:){5}[A-Fa-f0-9]{2})
IPV6 ((([0-9A-Fa-f]{1,4}:){7}([0-9A-Fa-f]{1,4}|:))|(([0-9A-Fa-f]{1,4}:){6}(:[0-9A-Fa-f]{1,4}|((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3})|:))|(([0-9A-Fa-f]{1,4}:){5}(((:[0-9A-Fa-f]{1,4}){1,2})|:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3})|:))|(([0-9A-Fa-f]{1,4}:){4}(((:[0-9A-Fa-f]{1,4}){1,3})|((:[0-9A-Fa-f]{1,4})?:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(([0-9A-Fa-f]{1,4}:){3}(((:[0-9A-Fa-f]{1,4}){1,4})|((:[0-9A-Fa-f]{1,4}){0,2}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(([0-9A-Fa-f]{1,4}:){2}(((:[0-9A-Fa-f]{1,4}){1,5})|((:[0-9A-Fa-f]{1,4}){0,3}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(([0-9A-Fa-f]{1,4}:){1}(((:[0-9A-Fa-f]{1,4}){1,6})|((:[0-9A-Fa-f]{1,4}){0,4}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(:(((:[0-9A-Fa-f]{1,4}){1,7})|((:[0-9A-Fa-f]{1,4}){0,5}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:)))(%.+)?
IPV4 (?<![0-9])(?:(?:25[0-5]|2[0-4][0-9]|[0-1]?[0-9]{1,2})[.](?:25[0-5]|2[0-4][0-9]|[0-1]?[0-9]{1,2})[.](?:25[0-5]|2[0-4][0-9]|[0-1]?[0-9]{1,2})[.](?:25[0-5]|2[0-4][0-9]|[0-1]?[0-9]{1,2}))(?![0-9])
IP (?:%{IPV6}|%{IPV4})
HOSTNAME \b(?:[0-9A-Za-z][0-9A-Za-z-]{0,62})(?:\.(?:[0-9A-Za-z][0-9A-Za-z-]{0,62}))*(\.?|\b)
HOST %{HOSTNAME}
IPORHOST (?:%{HOSTNAME}|%{IP})
HOSTPORT %{IPORHOST}:%{POSINT}

# paths
PATH (?:%{UNIXPATH}|%{WINPATH})
UNIXPATH (?>/(?>[\w_%!$@:.,-]+|\\.)*)+
TTY (?:/dev/(pts|tty([pq])?)(\w+)?/?(?:[0-9]+))
WINPATH (?>[A-Za-z]+:|\\)(?:\\[^\\?*]*)+
URIPROTO [A-Za-z]+(\+[A-Za-z+]+)?
URIHOST %{IPORHOST}(?::%{POSINT:port})?
# uripath comes loosely from RFC1738, but mostly from what Firefox
# doesn't turn into %XX
URIPATH (?:/[A-Za-z0-9$.+!*'(){},~:;=@#%_\-]*)+
#URIPARAM \?(?:[A-Za-z0-9]+(?:=(?:[^&]*))?(?:&(?:[A-Za-z0-9]+(?:=(?:[^&]*))?)?)*)?
URIPARAM \?[A-Za-z0-9$.+!*'|(){},~@#%&/=:;_?\-\[\]]*
URIPATHPARAM %{URIPATH}(?:%{URIPARAM})?
URI %{URIPROTO}://(?:%{USER}(?::[^@]*)?@)?(?:%{URIHOST})?(?:%{URIPATHPARAM})?

# Shortcuts
QS %{QUOTEDSTRING}

# Log formats
SYSLOGBASE %{SYSLOGTIMESTAMP:timestamp} (?:%{SYSLOGFACILITY} )?%{SYSLOGHOST:logsource} %{SYSLOGPROG}:
COMMONAPACHELOG %{IPORHOST:clientip} %{USER:ident} %{USER:auth} \[%{HTTPDATE:timestamp}\] "(?:%{WORD:verb} %{NOTSPACE:request}(?: HTTP/%{NUMBER:httpversion})?|%{DATA:rawrequest})" %{NUMBER:response} (?:%{NUMBER:bytes}|-)
COMBINEDAPACHELOG %{COMMONAPACHELOG} %{QS:referrer} %{QS:agent}