0
点赞
收藏
分享

微信扫一扫

ChatGPT在社交媒体聊天和评论分析中的应用如何?

彪悍的鼹鼠 2023-08-09 阅读 73
elk

目录

一、grok 正则捕获插件

自定义表达式调用

二、mutate 数据修改插件

示例:

●将字段old_field重命名为new_field

●添加字段

●将字段删除

●将filedName1字段数据类型转换成string类型,filedName2字段数据类型转换成float类型

●将filedName字段中所有"/“字符替换为”_"

●将filedName字段以"|"为分割符拆分数据成为数组

●合并 “filedName1” 和 “ filedName2” 两个字段

●用新值替换filedName字段的值

●添加字段first,值为message数组的第一个元素的值

●有条件的添加标签

三、multiline 多行合并插件

安装 multiline 插件

在线安装插件

离线安装插件

使用 multiline 插件

四、date 时间处理插件

时间戳详解:


一、grok 正则捕获插件

grok 使用文本片段切分的方式来切分日志事件

//内置正则表达式调用

%{SYNTAX:SEMANTIC}

SYNTAX代表匹配值的类型,例如,0.11可以NUMBER类型所匹配,10.222.22.25可以使用IP匹配。

SEMANTIC表示存储该值的一个变量声明,它会存储在elasticsearch当中方便kibana做字段搜索和统计,你可以将一个IP定义为客户端IP地址client_ip_address,如%{IP:client_ip_address},所匹配到的值就会存储到client_ip_address这个字段里边,类似数据库的列名,也可以把 event log 中的数字当成数字类型存储在一个指定的变量当中,比如响应时间http_response_time,假设event log record如下:

message: 192.168.110.100 GET /index.html 15824 0.043

可以使用如下grok pattern来匹配这种记录

%{IP:client_id_address} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:http_response_time}

在logstash conf.d文件夹下面创建filter conf文件,内容如下

vim /etc/logstash/conf.d/01-filter.conf
filter {
  grok {
    match => { "message" => "%{IP:client_id_address} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:http_response_time}" }
  }
}

以下是filter结果

client_id_address: 192.168.110.100
method: GET
request: /index.html
bytes: 15824
http_response_time: 0.043

logstash 官方也给了一些常用的常量来表达那些正则表达式,可以到这个 Github 地址查看有哪些常用的常量:

自定义表达式调用

举例:捕获10或11和长度的十六进制数的queue_id可以使用表达式(?<queue_id>[0-9A-F]{10,11})

message: 192.168.110.100 GET /index.html 15824 0.043
(?<remote_addr>\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}) (?<http_method>[A-Z]+) (?<request_uri>/.*) (?<response_bytes>[0-9]+) (?<response_time>[0-9\.]+)
filter {
  grok {
    match => { "message" => "(?<remote_addr>%{IP}) (?<http_method>[A-Z]+) (?<request_uri>/.*) (?<response_bytes>[0-9]+) (?<response_time>[0-9\.]+)"}
  }
}


如果表达式匹配失败,会生成一个tags字段,字段值为 _grokparsefailure,需要重新检查上边的match配置解析是否正确。

二、mutate 数据修改插件

它提供了丰富的基础类型数据处理能力。可以重命名,删除,替换和修改事件中的字段。

示例:

●将字段old_field重命名为new_field

filter {
    mutate {
        #写法1,使用中括号括起来
        rename => ["old_field" => "new_field"]

        #写法2,使用大括号{}括起来
        rename => { "old_field" => "new_field" }        
    }
}

●添加字段

filter {
    mutate {
        add_field => {
            "f1" => "field1"
            "f2" => "field2"
        }
    }
}

●将字段删除

filter {
    mutate {
        remove_field  =>  ["message", "@version", "tags"]
    }
}


●将filedName1字段数据类型转换成string类型,filedName2字段数据类型转换成float类型

filter {
    mutate {
        #写法1,使用中括号括起来
        convert  =>  ["filedName1", "string"]
        
        #写法2,使用大括号{}括起来
        convert => { "filedName2" => "float" }
    }
}

●将filedName字段中所有"/“字符替换为”_"

filter {
    mutate {
        gsub => ["filedName", "/" , "_"]
    }
}
●将filedName字段中所有",“字符后面添加空格
filter {
    mutate {
        gsub => ["filedName", "," , ", "]
    }
}

●将filedName字段以"|"为分割符拆分数据成为数组

filter {
    mutate {
        split => ["filedName", "|"]
    }
}

●合并 “filedName1” 和 “ filedName2” 两个字段

filter {
    merge  { "filedName2" => "filedName1" }
}

●用新值替换filedName字段的值

filter {
    mutate {
        replace => { "filedName" => "new_value" }
    }
}

●添加字段first,值为message数组的第一个元素的值

filter {
    mutate {
        split => ["message", "|"]
        add_field => {
            "first" => "%{[message][0]}"    
        } 
    }
}

●有条件的添加标签

filter {
    #在日志文件路径包含 access 的条件下添加标签内容
    if [path] =~ "access" {
        mutate {
            add_tag => ["Nginx Access Log"]
        }
    }

  #在日志文件路径是 /var/log/nginx/error.log 的条件下添加标签内容
    if [path] == "/var/log/nginx/error.log" {
        mutate {
            add_tag => ["Nginx Error Log"]
        }
    }
}

三、multiline 多行合并插件

java错误日志一般都是一条日志很多行的,会把堆栈信息打印出来,当经过 logstash 解析后,每一行都会当做一条记录存放到 ES, 那这种情况肯定是需要处理的。 这里就需要使用 multiline 插件,对属于同一个条日志的记录进行拼接。

2022-11-11 17:09:19.774[XNIo-1 task-1]ERROR com.passjava.controlle .NembercController-查询用户 活动数据失败,异常信息为:
    com.passjava.exception.MemberException: 当前没有配置活动规则
    at com.passjava.service.impL.queryAdmin(DailyServiceImpl.java:1444)
    at com.passjava.service.impl.dailyserviceImpL$$FastcLass
2022-11-11 17:10:56.256][KxNIo-1 task-1] ERROR com.passjava.controlle .NemberControl1er-查询员工 饭活动数据失败,异常信息为:
    com.passjava.exception.MemberException: 当前没有配置活动规则
    at com.passjava.service.impL.queryAdmin(DailyServiceImpl.java:1444)
    at com.passjava.service.impL.daiLyserviceImpL$$FastcLass

安装 multiline 插件

在线安装插件

cd /usr/share/logstash
bin/logstash-plugin install logstash-filter-multiline

离线安装插件

先在有网的机器上在线安装插件,然后打包,拷贝到服务器,执行安装命令

bin/logstash-plugin prepare-offline-pack --overwrite --output logstash-filter-multiline.zip logstash-filter-multiline
bin/logstash-plugin install file:///usr/share/logstash/logstash-filter-multiline.zip

检查下插件是否安装成功,可以执行以下命令查看插件列表

bin/logstash-plugin list

使用 multiline 插件

第一步:每一条日志的第一行开头都是一个时间,可以用时间的正则表达式匹配到第一行。
第二步:然后将后面每一行的日志与第一行合并。
第三步:当遇到某一行的开头是可以匹配正则表达式的时间的,就停止第一条日志的合并,开始合并第二条日志。
第四步:重复第二步和第三步。

filter {
  multiline {
    pattern => "^\d{4}-\d{1,2}-\d{1,2}\s\d{1,2}:\d{1,2}:\d{1,2}.\d{3}"
    negate => true
    what => "previous"
  }
}


●pattern:用来匹配文本的表达式,也可以是grok表达式

●what:如果pattern匹配成功的话,那么匹配行是归属于上一个事件,还是归属于下一个事件。previous: 归属于上一个事件,向上合并。next: 归属于下一个事件,向下合并

●negate:是否对pattern的结果取反。false:不取反,是默认值。true:取反。将多行事件扫描过程中的行匹配逻辑取反(如果pattern匹配失败,则认为当前行是多行事件的组成部分)

四、date 时间处理插件

用于分析字段中的日期,然后使用该日期或时间戳作为事件的logstash时间戳。

在Logstash产生了一个Event对象的时候,会给该Event设置一个时间,字段为“@timestamp”,同时,我们的日志内容一般也会有时间,但是这两个时间是不一样的,因为日志内容的时间是该日志打印出来的时间,而“@timestamp”字段的时间是input插件接收到了一条数据并创建Event的时间,所有一般来说的话“@timestamp”的时间要比日志内容的时间晚一点,因为Logstash监控数据变化,数据输入,创建Event导致的时间延迟。这两个时间都可以使用,具体要根据自己的需求来定。

filter {
    date {
        match => ["access_time", "dd/MMM/YYYY:HH:mm:ss Z", "UNIX", "yyyy-MM-dd HH:mm:ss", "dd-MMM-yyyy HH:mm:ss"]
        target => "@timestamp"
        timezone => "Asia/Shanghai"
    }
} 
  • match:用于配置具体的匹配内容规则,前半部分内容表示匹配实际日志当中的时间戳的名称,后半部分则用于匹配实际日志当中的时间戳格式,这个地方是整条配置的核心内容,如果此处规则匹配是无效的,则生成后的日志时间戳将会被input插件读取的时间替代。
  • 如果时间格式匹配失败,会生成一个tags字段,字段值为 _dateparsefailure,需要重新检查上边的match配置解析是否正确。
  • target:将匹配的时间戳存储到给定的目标字段中。如果未提供,则默认更新事件的@timestamp字段。
  • timezone:当需要配置的date里面没有时区信息,而且不是UTC时间,需要设置timezone参数。

时间戳详解:


//案例:
192.168.110.100 - - [07/Feb/2022:16:24:19 +0800] “GET /HTTP/1.1” 403 5039

现在我们想转换时间,那就要写出"dd/MMM/yyyy:HH:mm:ss Z"
你发现中间有三个M,你要是写出两个就不行了,因为两个大写的M表示两位数字的月份,可是我们要解析的文本中,月份则是使用简写的英文,所以只能去找三个M。还有最后为什么要加上个大写字母Z,因为要解析的文本中含有“+0800”时区偏移,因此我们要加上去,否则filter就不能正确解析文本数据,从而转换时间戳失败。

filter{
      grok{
           match => {"message" => ".* -\ -\ \[%{HTTPDATE:timestamp}\]"}
      }
      date{
           match => ["timestamp","dd/MMM/yyyy:HH:mm:ss Z"]
    }
}

运行结果:

{
          "host" => "localhost",
     "timestamp" => "07/Feb/2022:16:24:19 +0800",
    "@timestamp" => 2022-02-07T08:24:19.000Z,
       "message" => "192.168.80.10 - - [07/Feb/2022:16:24:19 +0800] \"GET /HTTP/1.1\" 403 5039",
      "@version" => "1"
}

在上面那段rubydebug编码格式的输出中,@timestamp字段虽然已经获取了timestamp字段的时间,但是仍然比北京时间晚了8个小时,这是因为在Elasticsearch内部,对时间类型字段都是统一采用UTC时间,而日志统一采用UTC时间存储,是国际安全、运维界的一个共识。其实这并不影响什么,因为ELK已经给出了解决方案,那就是在Kibana平台上,程序会自动读取浏览器的当前时区,然后在web页面自动将UTC时间转换为当前时区的时间。

举报

相关推荐

0 条评论