参考网址：官方文档、logstash示例

Logstash是一个开源数据收集引擎，具有实时管道功能。可以动态地将来自不同数据源的数据统一起来，并将数据标准化到你所选择的目的地。Logstash 是一个接收、处理、转发日志的工具，支持系统日志、webserver 日志、错误日志、应用日志，总之包括所有可以抛出来的日志类型。在一个典型的使用场景下(ELK)：用 Elasticsearch 作为后台数据的存储，kibana用来前端的报表展示。Logstash 在其过程中担任搬运工的角色，它为数据存储、报表查询和日志解析创建了一个功能强大的管道链。Logstash 提供了多种多样的 input，filters，codecs 和 output 组件，让使用者轻松实现强大的功能。Logstash 收集数据对象就是日志文件，使用 Logstash 对日志文件进行收集和统一过滤，变成可读性高的内容，方便开发者或运维人员观察，从而有效的分析系统或项目运行的性能，做好监控和预警的准备工作等。

1.工作原理

Logstash 通过管道进行运作，管道有两个必需的元素，输入和输出，还有一个可选的元素，过滤器。输入插件从数据源获取数据，过滤器插件根据用户指定的数据格式修改数据，输出插件则将数据写入到目的地。

image

先了解一个概念：事件。Logstash 每读取一次数据的行为叫做事件。

Logstash管道中的每个输入阶段都在自己的线程中运行。输入将事件写入位于内存(默认)或磁盘上的中心队列。每个管道工作线程从队列中取出一批事件，通过配置的过滤器运行这批事件，然后输出经过过滤的事件。批处理的大小和管道工作线程的数量是可配置的。

默认情况下,Logstash使用内存有限队列之间的管道(输入→过滤器和过滤器→输出)缓冲事件。如果Logstash不安全终止，那么存储在内存中的任何事件都将丢失。为了防止数据丢失，您可以启用Logstash将正在运行的事件持久化到磁盘。

Logstash 工作的三个阶段：input => filter => output

input ：数据输入端，可以接收来自任何地方的源数据。

• file：从文件中读取
• syslog：监听在514端口的系统日志信息，并解析成RFC3164格式。
• kafka：从kakfka topic 中获取数据
• beat：接收来自Filebeat的事件

Filter ：数据中转层，主要进行格式处理，数据类型转换、数据过滤、字段添加，修改等，常用的过滤器如下。

• grok: 通过正则解析和结构化任何文本。Grok 目前是logstash最好的方式对非结构化日志数据解析成结构化和可查询化。logstash内置了120个匹配模式，https://github.com/logstash-plugins/logstash-patterns-core/tree/master/patterns，满足大部分需求。

• mutate: 在事件字段执行一般的转换。可以重命名、删除、替换和修改事件字段。

• drop: 完全丢弃事件，如debug事件。

• clone: 复制事件，可能添加或者删除字段。

• geoip: 添加有关IP地址地理位置信息。

output :是logstash工作的最后一个阶段，负责将数据输出到指定位置，兼容大多数应用，常用的有:

• elasticsearch: 发送事件数据到 Elasticsearch，便于查询，分析，绘图。

• file: 将事件数据写入到磁盘文件上。

• kafka: 将事件数据发送至高性能kafka分布式系统，便于临时存储，查询，分析，大数据消费场景。

• redis:将数据发送至redis-server，常用于中间层暂时缓存。

• influxdb: 发送指标数据到influxdb。

• statsd: 发送事件数据到 statsd。

2.Logstash配置

2.1主配置

logstash.yml:

node.name：节点的描述性名称。默认值为机器的主机名

path.data：Logstash及其插件用于满足任何持久需求的目录，默认值LOGSTASH_HOME/data。

pipeline.id： 管道的ID，默认值为main，可以更改；

pipeline.workers:
并行执行管道过滤器和输出阶段的工作器数量。如果发现事件正在备份，或者CPU未饱和，请考虑增加此数量以更好地利用机器处理能力，默认值为主机的CPU核心数

pipeline.batch.size：
在尝试执行其过滤器和输出之前，单个工作线程将从输入收集的最大事件数。较大的批量通常更有效，但代价是增加了内存开销，需要在jvm.options配置文件中增加JVM堆空间，默认值为125。

pipeline.batch.delay：
创建管道事件批处理时，在将小型批处理分派给管道工作者之前等待每个事件的时间以毫秒为单位，默认值为50

pipeline.unsafe_shutdown：
设置true为时，强制Logstash在关闭期间退出，即使内存中仍有事件。默认情况下，Logstash将拒绝退出，直到所有已接收的事件都被推送到输出。启用此选项可能会导致关闭期间数据丢失，默认值为false

path.config：
主管道的Logstash配置的路径。如果指定目录或通配符，则会按字母顺序从目录中读取配置文件。

config.test_and_exit：
设置为时true，检查配置是否有效，然后退出。使用此设置不会检查grok模式的正确性。Logstash可以从目录中读取多个配置文件，默认值为false

config.reload.automatic：
设置true为时，定期检查配置是否已更改，并在配置发生更改时重新加载配置。这也可以通过SIGHUP信号手动触发，默认值为false

config.reload.interval：几秒钟内，Logstash会检查配置文件中的更改，默认值为3s

config.debug：
设置为时true，将完全编译的配置显示为调试日志消息。你还必须设置log.level:
debug。警告：日志消息将包含作为纯文本传递给插件配置的任何密码选项，并可能导致明文密码出现在您的日志中，默认值为false

config.support_escapes
设置true为时，带引号的字符串将处理以下转义序列：\n成为文字换行符（ASCII 10）。\r成为文字回车（ASCII13）。\t成为文字标签（ASCII 9）。\\成为一个字面反斜杠\。\"成为字面双引号。\'成为字面引号，默认值为false

modules
配置时，modules必须在此表中描述的嵌套YAML结构中。

queue.type：
用于事件缓冲的内部排队模型。指定memory基于内存的传统队列，或persisted基于磁盘的ACKed队列（持久队列），默认值为memory，推荐使用持久队列

path.queue：
启用持久队列时将存储数据文件的目录路径（queue.type: persisted）。

queue.page_capacity：
启用持久队列时使用的页面数据文件的大小（queue.type: persisted）。队列数据由分成页面的仅附加数据文件组成，默认值为64MB

queue.max_events：
启用持久队列时队列中未读事件的最大数量（queue.type: persisted），默认值为0（无限制）

queue.max_bytes：
队列的总容量，以字节数表示。确保磁盘驱动器的容量大于此处指定的值。如果同时指定了两者queue.max_events，queue.max_bytes则Logstash将使用先达到的标准，默认值为1024mb

queue.checkpoint.acks：
启用持久队列时强制检查点之前的最大ACK事件数（queue.type: persisted）。指定queue.checkpoint.acks: 0将此值设置为无限制，默认值为1024

queue.checkpoint.writes：
启用持久队列时强制检查点之前写入事件的最大数量（queue.type: persisted）。指定queue.checkpoint.writes:

0将此值设置为无限制，默认值为1024

queue.checkpoint.retry：
启用后，对于任何失败的检查点写入，Logstash将针对每次尝试检查点写入重试一次。不会重试任何后续错误。这是仅在具有非标准行为（如SAN）的文件系统上看到的失败检查点写入的解决方法，除特殊情况外，不建议这样做，默认值为false

dead_letter_queue.enable：
用于指示Logstash启用插件支持的DLQ功能的标志，默认值为false

dead_letter_queue.max_bytes：
每个死信队列的最大大小。如果条目超过此设置会增加死信队列的大小，则会删除条目。默认值为1024mb

path.dead_letter_queue：
将为死信队列存储数据文件的目录路径。

http.host：绑定地址，默认值为"127.0.0.1"

http.port：绑定端口，默认值为9600

log.level：日志级别。有效选项包括：fatal，error，warn，info，debug，trace，info

log.format：日志格式。设置为json或plain，默认值为plain

path.logs：Logstash将其日志写入的目录，默认值为LOGSTASH_HOME/logs

path.plugins：
哪里可以找到自定义插件。您可以多次指定此设置以包含多个路径。插件预计将在一个特定的目录层次结构： PATH/logstash/TYPE/NAME.rb其中TYPE是inputs，filters，outputs，或codecs，并且NAME是插件的名称。

2.2Input配置

Beat：从beat采集

input {
  beats {
    port => 5044
  }
}

File：从文件采集

input {
  file {
    path => "/var/log/*/*.log"
  }
}

Elasticsearch：从es集群采集数据

input {
  elasticsearch {
    hosts => "localhost"
    query => '{ "query": {"match": { "statuscode": 200 } }, "sort": ["_doc" ] }'
  }
}

Kafka：从kafka获取数据

input {
  kafka {
    bootstrap_servers =>"10.22.67.5:9992 "
    topics => ["test"]
    codec => "json"
    security_protocol =>"SASL_PLAINTEXT"
    sasl_mechanism => "PLAIN"
    jaas_path =>"/etc/logstash/kafka-client-jaas.conf"
    consumer_threads => 2
    group_id => "logstash-bigdata"
    auto_offset_reset => "latest"
    max_poll_records => "500"
    max_poll_interval_ms =>"30000"
    session_timeout_ms => "30000"
    request_timeout_ms => "60000"
    auto_commit_interval_ms =>"5000"
    check_crcs => "false"
    heartbeat_interval_ms =>"9000"
    partition_assignment_strategy =>"org.apache.kafka.clients.consumer.RoundRobinAssignor"
    }
}

Jdbc：从数据库采集数据

input {
  jdbc {
    jdbc_driver_library =>"mysql-connector-java-5.1.36-bin.jar"
    jdbc_driver_class =>"com.mysql.jdbc.Driver"
    jdbc_connection_string =>"jdbc:mysql://localhost:3306/mydb"
    jdbc_user => "mysql"
    parameters => {"favorite_artist" => "Beethoven" }
    schedule => "* * * * *"
    statement => "SELECT * from songswhere artist = :favorite_artist"
  }
}

2.3Output配置

Elasticsearch：输出至ES集群

output {
if [log_type] == "tomcat" and([department] == "zxbi" or [department] == "jfbi") {
  elasticsearch {
    hosts =>["https://rjjd-node01:9200"]
    index =>"jfbi.prd-%{+YYYY.MM}"
    user => "beatuser"
    password =>"1iQ7w3LQlOhHR6Rg1iQ7w3L"
    http_compression => "true"
    sniffing => "false"
    ssl => true
    ssl_certificate_verification =>"true"
    cacert => "/etc/ssl/xxx_ca.crt"
    id => "zxbi"
  }
 }
}

Influxdb：输出至influxdb数据库

output {
  influxdb {
  host => "10.1.2.2:8086"
  data_points => “{ }”
  }
}

File：输出至文件

output {
file {
  path => “/tmp/stdout.log”
  codec => line { format => "customformat: %{message}"}
  }
}

Opentsdb：输出至opentsdb

output {
  opentsdb {
  host => "localhost"
  metrics => [“%{host}”, “%{hostname}”]
  port => “4242”
  }
}

Stdout：输出至控制台

output {
  stdout { codec => json }
}

3.Logstash解析

最常用的解析插件：grok、date、drop、geoip、json、kv、ruby、split、mutate

3.1Grok

grok匹配模式语法为：%{SYNTAX:SEMANTIC:TYPE}

• SYNTAX: 正则表达式、预定义的正则表达式名称
• SEMANTIC: 标识符，标识匹配后的数据
• TYPE: 可选的类型，目前支持int、float
  例如：NUMBER可以匹配3.44，IP可以匹配：192.168.21.2

一个简单的日志格式如下：
192.168.21.2 GET /index.html 15823 0.023
grok匹配模式可以为：
${IP:client} %{WORD:method}%{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}
假设该条日志可能来自一个文件：

input {
  file {
    path => "/var/log/http.log"
  }
}

filter {
  grok {
    match => { "message" =>"%{IP:client} %{WORD:method} %{URIPATHPARAM:request}%{NUMBER:bytes} %{NUMBER:duration}" }
  }
}

在grok过滤后，可以得到额外一下字段：
client: 192.168.21.2
method: GET
request: /index.html
bytes: 15823
duration: 0.023

grok表达式支持的可选参数：

• break_on_match：默认为true代表匹配成功后则不匹配后面的表达式，false相反
• keep_empty_captures：默认为false，字段没有值时则不输出字段，true则相反
• match：表达式模式
• overwrite：字段重新命名
• patterns_dir：表达式文件夹
• patterns_files_glob：表达式文件夹正则匹配
• tag_on_failure：匹配失败后添加tag
• tag_on_timeout：匹配超时添加tag
• add_field：自定义添加字段
• add_tag： 自定义添加tag
• enable_metric：启用监控
• id：表达式id
• remove_field：移除字段
• remove_tag：移除tag

3.2Date

日期过滤器用于解析字段中的日期，然后使用该日期或时间戳作为事件的logstash时间戳。

filter {
  date {
  match => [ "logdate", "MMM dd yyyyHH:mm:ss" ]
  target => “localtime”
  }
}

支持的参数：

Locale：日期语言环境
Match：表达式
Timezone：时间时区
Tag_on_failure：匹配失败后添加tag
Target：匹配成功后目标字段

时间匹配参数：

时间匹配练习：
2019-08 12:22:22 239 ---yyyy-MM HH:mm:ss SSS
Jul 2 2019 22:30 ---MMM d yyyy HH:mm
12 Mar 2019 22:30:30 +08:00 ---dd MMM yyyy HH:mm:ss ZZ

3.3Drop

当不需要某些数据的时候，可以使用drop插件丢弃，例如：

filter {
  if [loglevel] == "debug" {
    drop { }
  }
}

3.4Geoip

GeoIP过滤器根据Maxmind
GeoLite2数据库中的数据添加有关IP地址的地理位置的信息。
配置表达式：

filter {
  geoip {
    database =>“/tmp/xx.db”
    fields => [“ip”,”longitute”,”city_name”]
    source => “ip”
    target => “geoip”
  }
}

3.5Json

默认情况下，它会将解析后的JSON放在Logstash事件的根（顶层）中，但可以使用配置将此过滤器配置为将JSON放入任意任意事件字段 target。

当在解析事件期间发生不良事件时，此插件有一些回退场景。如果JSON解析在数据上失败，则事件将不受影响，并将标记为 _jsonparsefailure; 然后，您可以使用条件来清理数据。您可以使用该tag_on_failure选项配置此标记。

如果解析的数据包含@timestamp字段，则插件将尝试将其用于事件@timestamp，如果解析失败，则字段将重命名为，_@timestamp并且事件将使用a标记 _timestampparsefailure。

filter {
  json {
    source => "message"
  }
}

3.6Kv

此过滤器有助于自动解析各种消息（或特定事件字段）类似foo=bar。

例如，如果您有一条包含的日志消息ip=1.2.3.4 error=REFUSED，则可以通过配置来自动解析这些消息。

filter {
  kv {
    default_keys => [“host”,”ip”]
    exclude_keys => [“times”]
    field_split => “&?”
  }
}

支持的参数：
allow_duplicate_values
default_keys
exclude_keys
field_split
field_split_pattern
include_brackets
include_keys
prefix
recursive
remove_char_key
remove_char_value
source
target
tag_on_failure
tag_on_timeout
timeout_millis
transform_key
transform_value
trim_key
trim_value
value_split
value_split_pattern
whitespace

3.7Ruby

执行ruby代码。此过滤器接受内联ruby代码或ruby文件。这两个选项是互斥的，具有稍微不同的工作方式。

例如：

filter {
  ruby {
    code => "event.cancel if rand <=0.90"
  }
}

3.8Split

拆分筛选器通过拆分其中一个字段并将拆分产生的每个值放入原始事件的克隆来克隆事件。要拆分的字段可以是字符串或数组。

此过滤器的一个示例用例是从exec输入插件获取输出，该插件为命令的整个输出发出一个事件，并按换行分割该输出 - 使每一行成为事件。

拆分过滤器还可用于将事件中的数组字段拆分为单个事件。JSON和XML中一种非常常见的模式是利用列表将数据组合在一起。

filter {
  split {
    field => "results"
    target=> “messages”
    terminator => “\n”
  }
}

3.9Mutate

强大的mutate过滤器，可以对数据进行增删改查。支持的语法多，且效率高

按照执行顺序排列：
coerce
rename：重命名字段
update：更新数据
replace：替换字段值
convert：转换字段类型
gsub：替换字符
uppercase：转为大写的字符串
capitalize：转换大写字符串
lowercase：转为小写的字符串
strip：剥离字符空白
remove：移除字段
split：分离字段
join：合并数组
merge：合并多个数组
copy：复制字段

例如：

filter {
  mutate {
    split => ["hostname","."]
    add_field => {"shortHostname" => "%{hostname[0]}" }
  }
  mutate {
    rename =>["shortHostname", "hostname" ]
  }
}

4.Logstash性能优化

Logstash本质是一个数据处理流程，因此从建模的角度来看，最好的架构就是流水线工作，而刚好logstash官方设计也是基于流水线工作设计。

Logstash单实例本质是一个处理单元，因此最好的架构就是网状架构，每个实例加载多个管道，每个管道可以自由的被多个处理单元进行处理数据。

logstash先对单个实例进行性能最大化，然后将其加到处理流水线中，由集群管理中心对其进行处理事务分配，从logstash主配置文件中我们知道，单个logstash的性能最大化由3个元素组成，batch.size（批处理大小），workers（工作线程数），batch.delay（批处理频率），单个实例每秒最大tps计算公式=批处理大小工作线程数/处理频率，加入1台8核16G内存的实例，它的配置batchsize为1千，工作线程数为8，频率为200ms，它所能接受的理论tps为=1千8/0.2=4万，但实际上肯定达不到这个数值，因为这中间需要jvm内存的缓存，而这也是logstash常常需要优化的部分。

假设4万条数据，每条数据为20KB，那么每秒钟需要的数据容量=4万20KB8=6.4gb，那么在你的jvm里面需要常驻6.4gb的堆空间来加载这些数据，同时处理过程中需要有3个阶段，因此实际使用过程中需要的对空间需要大于6.4gb，那么我们如何去分析logstash的性能呢？

image

image

在第一个图中，我们看到CPU没有得到非常有效的使用。实际上，JVM经常需要停止VM以获得“完整的GC”。完全垃圾收集是过度记忆压力的常见症状。这在CPU图表上的峰值中可见。在更有效配置的示例中，GC图形模式更平滑，并且CPU以更均匀的方式使用。您还可以看到在分配的堆大小和允许的最大大小之间有足够的空间，这为JVM GC提供了很大的工作空间。与资源密集程度较高的旧Gen“Full”GC所花费的时间相比，过度分配的VM在高效的Eden GC中花费的时间非常少。在一般实践中，我们需要让jvm的堆内存实际使用大小尽可能接近最大大小，且CPU处于比较平衡的趋势在处理事务。

问题：在项目实践过程中，我们的logstash单个实例应该配置多大的资源？

完成了单个实例的性能优化之后，对于超大流量处理，我们如何使用多个实例进行并发处理呢？使用更多实例还是更多管道，谁的机制更好呢？这里又带来了一个新的概念，logstash的管道机制。

logstash集群可以使用负载均衡模式从kafka集群中快速的获取到最大数据量。例如同一台物理机，配置为8核16G，使用1个8核16G资源实例同时对5个topic的5个分区进行处理，跟4个2核4G资源实例对5个topic的5个分区进行处理，谁会更快？

Logstash提供了一种通过调用的配置文件来执行此操作的方法pipelines.yml。此文件必须放在path.settings文件夹中并遵循以下结构：

• pipeline.id:my-pipeline_1

  path.config:"/etc/path/to/p1.config"

  pipeline.workers: 3

• pipeline.id:my-other-pipeline

  path.config:"/etc/different/path/p2.cfg"

  queue.type: persisted

此文件以YAML格式化，并包含字典列表，其中每个字典描述一个管道，每个键/值对指定该管道的设置。该示例显示了由ID和配置路径描述的两个不同管道。对于第一个管道，值pipeline.workers设置为3，而在另一个管道中，启用持久队列功能。未在pipelines.yml文件中显式设置的设置值将回退到logstash.yml设置文件中指定的默认值。

使用管道注意方法：

当启动没有参数的Logstash时，它将读取pipelines.yml文件并实例化文件中指定的所有管道。另一方面，当使用-e或时-f，Logstash会忽略该pipelines.yml文件并记录有关它的警告。

如果当前配置的事件流不共享相同的输入/过滤器和输出，并且使用标签和条件将彼此分开，则使用多个管道显得特别有用。（问题：猜想一下有用的地方会体现在哪些方面）

在单个实例中具有多个管道还允许这些事件流具有不同的性能和持久性参数（例如，管道工作者和持久队列的不同设置）。这种分离意味着一个管道中的阻塞输出不会在另一个管道中施加背压。

也就是说，考虑到为单个管道调整默认值，考虑管道之间的资源竞争非常重要。因此例如考虑减少每个管道使用的管道工作者的数量，因为默认情况下每个管道将使用每个CPU核心1个工作线程。

持久队列和死信队列是按管道隔离的，其位置由pipeline.id值命名。

5.启动和停止

5.1开机自启动

确认主机运行的logstash实例及其参数

把所有实例的启动命令追加到/etc/rc.local

例如：在/etc/rc.local追加（注意绝对路径）

nohup /hadoop/logstash-${version}/bin/logstash -f /hadoop/logstash-${version}/conf/test.conf &>/var/log/logstash/test.log &

5.2启动logstash

Logstash均需要配置成开机自启动模式。

cd LOGSTASH_HOME
./bin/logstash -f config/test.conf

后台启动：

nohup./bin/logstash -f config/test.conf -w 16 1>/var/log/logstash/{系统名} 2>&1 &
启动多个logstash配置，新建一个目录（conf）存放多个logstash的配置文件，
./bin/logstash -f conf/*

5.3停止logstash

ps -ef |grep logstash
kill -9 $pid

6.实践应用

百看不如一做，做几个小题目，走了不少弯路才绕出来，悲催。。。一开始不熟悉会走很多迷途，且行且珍惜。我自己更才发现官方文档原来是特别有用，要仔细看看的，grok里的匹配也很强大，回头另外总结下grok匹配规则，那下面来几个小题目练习下（此处非标准答案，因己解析哈）：

Jul 13 09:01:01 localhost systemd: Stopping user-0.slice.
Jul 13 09:03:56 localhost systemd: Created slice user-0.slice.
Jul 13 09:03:56 localhost systemd: Starting user-0.slice.
解析说明：此处的难点在于systemd:后的内容，有空格就不能使用空格split；时间格式就要仔细研究下grok的规则了

input {
  stdin {}
}
filter {
  grok {
    match => { "message" => "%{SYSLOGTIMESTAMP:logtime} %{IPORHOST:hostname} %{GREEDYDATA:mess}" }
  }
  mutate {
    split => ["mess",": "]
    add_field => { "%{[mess][0]}" => "%{[mess][1]}" }
    remove_field => ["mess","@version"]
  }
}
output {
    stdout { codec => rubydebug }
}

1.png

type=CRED_REFR msg=audit(1563023036.263:1242): pid=6977 uid=0 auid=0 ses=156 subj=unconfined_u:unconfined_r:unconfined_t:s0-s0:c0.c1023 msg='op=PAM:setcred grantors=pam_unix acct="root" exe="/usr/sbin/sshd" hostname=securenat-daf2680060a6 addr=192.168.2.63 terminal=ssh res=success'
解析说明：此处出现了多层嵌套，而且msg字段有两处（我自己理解的哈），最初开始使用的不是gsub，用split有点麻烦，用gsub直接替换干掉，干脆利索（我也是最后才发现“宝藏”）；还有一个就是那个add_field用得有点醉了，纠结就纠结在那个audit放在msg字段里面

input {
  stdin {}
}
filter {
  kv {}
  mutate{
    gsub => ["[msg][0]","\(","=" ]
    gsub => ["[msg][0]","\):","" ]
  }
  kv{
    source => "[msg][0]"
  }
  kv{
    source => "[msg][1]"
    trim_value => "\"'"
    target => "msg"
  }
  mutate {
    add_field => { "[msg][audit]" => "%{[audit]}" }
    remove_field => ["audit","@version"]
  }
}
output {
    stdout { codec => rubydebug }
}

2.png

2019-06-19 08:55:28,818 [0cfad026-4b3e-4bd5-90c3-d993d67426571560851187560] [d95ddac9-8525-463a-b63f-f31984d745d4] [http-nio-8083-exec-15] INFO c.e.c.filter.ParameterWrapperFilter - 请求业务参数：{
"deviceId" : "hedc76d5b9f6b8b374705b7684f2f5e13",
"appVersion" : "4.4.0",
"osType" : "iOS",
"osVersion" : "OS12.2",
"serviceId" : "https://sit-mobile.essence.com.cn/self-select-stock/public/userStockSign/queryStock",
"appId" : "mobile.essence.com.cn",
"reqdata" : {
"accessToken" : "eyJhbGciOiJIUzUxMiJ9.eyJqdGkiOiIwZDYxNjg1MC0zNDM5LTQ4MDgtOTliOS1hYmQyMzdkOTk1ZTAiLCJpc3MiOiJhZG1pbkBlc3NlbmNlLmNvbS5jbiIsImlhdCI6MTU2txUZA",
"md5Sign" : ""
},
"requestId" : "1560905683118145",
"requestTime" : "2019-06-19 08:54:43"
}
解析说明：此处我一上来就使用grok匹配，思路就错了点，多行那个应该是首先考虑的的，“请求业务参数”那里是中文冒号，grok是可以配置的，然后后面的多行直接使用json，手起刀落，一次性搞定；另一种方法里的奥妙在于取字符串的“(?<info6>(?<=请求业务参数：)(.*)/?)”

input {
  file {
    path => ["/zork/logstash-7.2.0/conf/test3.log"]
    start_position => "beginning"
    sincedb_path => "/dev/null"
    codec => multiline {
      pattern => "^%{TIMESTAMP_ISO8601}"
      negate => true 
      what => "previous"
    }
  }
}

filter {
  grok {
    match => {
      "message" => "%{TIMESTAMP_ISO8601:timestamp} \[%{DATA:info1}\] \[%{DATA:info2}\] \[%{DATA:info3}\] %{LOGLEVEL:loglevel}  %{DATA:info4} - %{DATA:info5}：%{GREEDYDATA:info6}"
    }
  }
  date {
    match => [ "timestamp","yyyy-MM-dd HH:mm:ss,SSS" ]
  }
  json {
    source => "info6"
    target => "请求业务参数"
    remove_field => [ "info5","info6","tags","@version" ]
  }
}
output {
    stdout { codec => rubydebug }
}

或者

input {
  file {
    path => ["/zork/logstash-7.2.0/conf/test3.log"]
    start_position => "beginning"
    sincedb_path => "/dev/null"
    codec => multiline {
      pattern => "^\s*"
      negate => false
      what => "previous"
    }
  }
}

filter {
  grok {
    match => {
      "message" => "%{TIMESTAMP_ISO8601:timestamp} \[%{DATA:info1}\] \[%{DATA:info2}\] \[%{DATA:info3}\] %{LOGLEVEL:loglevel}  %{DATA:info4} - %{GREEDYDATA:info5}"
    }
  }
  date {
    match => [ "timestamp","yyyy-MM-dd HH:mm:ss,SSS" ]
  }
  grok {
    match => { "info5" => "(?<info6>(?<=请求业务参数：)(.*)/?)" }
  }
  json {
    source => "info6"
    target => "请求业务参数"
    remove_field => [ "info5","info6","tags","@version" ]
  }
}
output {
    stdout { codec => rubydebug }
}

3.png

06-19 2019 09:09:03,365 [43371440-7c77-4f67-b27e-33d3347bea2e1560906497865] [7f03a62f-92e8-4e3c-bf8b-8af6e3460121] [http-nio-8083-exec-17] INFO c.e.o.stock.repository.UserCenterApi - UC103009,名称=>自选股查询 ,业务参数=>{USER_ID=1597929340065856}
解析说明：此处的点在于这个时间，反正我是没找到grok里匹配这个时间的模式，也绕了一大圈圈

input {
  stdin {}
}

filter {
  grok {
    match => {
      "message" => "%{MONTHNUM:month}-%{MONTHDAY:day} %{YEAR:year} %{HOUR:hour}:%{MINUTE:minute}:%{SECOND:second} \[%{DATA:info1}\] \[%{DATA:info2}\] \[%{DATA:info3}\] %{LOGLEVEL:loglevel}  %{DATA:info4} - %{DATA:info5},%{GREEDYDATA:info6}"
    }
  }
  ruby{
    code => "
      timestamp = event.get('month')+'-'+event.get('day')+' '+event.get('year')+' '+event.get('hour')+':'+event.get('minute')+':'+event.get('second')
      event.set('timestamp',timestamp)
    "
    remove_field => [ "month","day","year","hour","minute","second","@version"]
  }
  kv {
    source => "info6"
    field_split => " ,"
#    value_split => "=>"
    trim_value => ">"
  }
  kv {
    source => "[业务参数]"
    trim_key => "{}"
    trim_value => "{}"
    target => "业务参数"
  }
}
output {
    stdout { codec => rubydebug }
}

4.png