目 录CONTENT

文章目录

ELK相关笔记05-logstash的使用

cplinux98
2022-11-10 / 0 评论 / 0 点赞 / 1,012 阅读 / 1,679 字 / 正在检测是否收录...

00:文章简介

通过3个实践案例,来介绍logstash。

01:实践1:接受filebeat传输的json格式数据,并在当前屏幕上展示效果

1.1:属性解析

input设置从哪读取数据
    beats设置从beats组件中接受数据
    codec用于传入数据的编解码,默认是plain-单行字符串,设置为json就使用json解析
output设置从哪输出数据
    stdout输出到标准输出(控制台)
    rubydebug设置输出格式为debug信息

1.2:设置filebeat配置文件

# 备份配置文件
cp filebeat.yml backup/filebeat.bak-es
# 编辑配置文件
vim filebeat.yml
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/nginx/access.log
  fields:
    log_type: "access"
  fields_under_root: true
  json.keys_under_root: true
  json.overwrite_keys: true

- type: log
  enabled: true
  paths:
    - /var/log/nginx/error.log
  fields:
    log_type: "error"
  fields_under_root: true

output.logstash:
  hosts: ["10.0.0.112:5044"]

# 重启服务
systemctl restart filebeat.service

1.3:设置logstash配置文件

# backup conf
cd /etc/logstash/conf.d/
cp logstash.conf ../conf.bak/logstash.bak-20210825
# edit conf
vim logstash.conf 

input {
  beats {
    port => 5044
    codec => json
  }
}

output {
  stdout { codec => rubydebug }
}


# 停止服务
systemctl stop logstash.service

1.4:查看效果

logstash -f logstash.conf

# 其他节点访问
curl 10.0.0.114

02:实践2:将收集到的不同格式数据,传输到不同的es索引中

2.1:属性解析

input设置从哪读取数据
    beats设置从beats组件中接受数据
    codec用于传入数据的编解码,默认是plain-单行字符串,设置为json就使用json解析
output设置从哪输出数据
    elasticsearch设置输出到es

2.2:配置logstash配置文件

# CTRL+c结束上个终端

# 备份配置文件
cp logstash.conf ../conf.bak/logstash.bak-stdout

# 修改配置文件
vim logstash.conf 

input {
  beats {
    port => 5044
    codec => json
  }
}

output {
  if [log_type] == "access" {
    elasticsearch {
      hosts => ["10.0.0.110:9200"]
      index => "nginx-access-%{+yyy.MM.dd}"
    }
  }

  if [log_type] == "error" {
    elasticsearch {
      hosts => ["10.0.0.110:9200"]
      index => "nginx-error-%{+yyyy.MM.dd}"
    }
  }
}



# 启动logstash
logstash -f logstash.conf

2.3:查看效果

# 制作访问日志
curl 10.0.0.114
# 制作错误日志 filebeat节点
echo "error" /var/log/nginx/error.log

03:实践3:通过geo.ip获取客户端坐标

3.1:属性解析

对数据进行分析的方式无非就是图和表,所谓表的分析,就是将数据统计后,按照顺序排列起来,比如我
们平常所说的TOPn、排名前5客户端IP、排名前5用户访问页面等。如果我们想看看我们的网站在全国各地的
分布热度,表虽然能够在一定程度上分析,但是理解起来不太方面,不能做到一目了然,而图就可以达到对应
的效果。
接下来我们就需要实现将我们项目的客户端IP在全国地图展示出来,这个功能我们就需要一个IP数据库
来识别客户端IP所在的地理位置。

# ip数据库
GeoLite2(https://dev.maxmind.com/geoip/geoip2/geolite2/)是一个解压直接使用的城市
IP地理位置数据库,支持IPv4和IPv6两种地址,GeoLite2的免费和收费版本,仅仅是精确度的差别,我们
教学用的就是最新的免费版本。据说每周二都会更新IP地址数据。
GeoIP根据经度和纬度生成 [geoip][location] 格式数据。该字段以GeoJSON格式存储。借助于
Elasticsearch模板输出该格式字段添加到Elasticsearch索引的geo_point(GeoJSON格式即
geoip),从而为其他应用程序提供GeoJSON的地理区域服务。

# geojson
我们es中存储的数据是是json格式,属性值是单字符串格式,而IP数据库中的解析出来的经纬度也是浮
点型数字,而地图的位置标识以数组格式存在的,所以我们将输入到es之前,还需要对logstash接收的数据
进行定制修改,这就涉及到logstash的filter组件定制数据格式[数据类型(浮点型)+数据组合(数组)]。

# es模板示例
es模板示例:
https://www.elastic.co/guide/en/logstash/current/plugins-output-selasticsearch.html

# 依赖的插件
logstash对于IP数据库的识别主要靠如下几个插件的功能
# logstash-plugin list --verbose | egrep 'geoip|mutate|useragent'
logstash-filter-geoip (7.2.2)
logstash-filter-mutate (3.5.2)
logstash-filter-useragent (3.3.1)
插件解析:
useragent 从数据中获取用户相关属性信息
geoip 根据用户IP信息,获取具体地理位置信息
mutate 对数据进行格式转换

# 定制数据格式
定制数据格式
filter {
    geoip {
        source => "客户端IP字段名" # 根据用户行为数据获取客户端IP地址
        add_field => { "新字段名称" => "属性值" } # 增加扩展字段,主要是数组格式的经纬度数据
    }
    mutate {
        convert => [ "属性", "数据格式"] # 将普通字符串数据,转换为特点浮点类型
    }
}   

# 地址数据库
scp /usr/share/elasticsearch/modules/ingest-geoip/GeoLite2-City.mmdb root@10.0.0.112:/etc/logstash/

3.2:配置文件

cp logstash.conf ../conf.bak/logstash.bak-20210826
root@Logstash:/etc/logstash/conf.d# cat logstash.conf 
# Sample Logstash configuration for creating a simple
# Beats -> Logstash -> Elasticsearch pipeline.

input {
  beats {
    port => 5044
    codec => json
  }
}

filter {
  if [log_type] == "access" {
    geoip {
      source => "http_x_real_ip"
      target => "geoip"
      database => "/etc/logstash/GeoLite2-City.mmdb"
      add_field => [ "[geoip][coordinates]", "%{[geoip][longitude]}" ]
      add_field => [ "[geoip][coordinates]", "%{[geoip][latitude]}" ]
    }
    mutate {
      convert => [ "[geoip][coordinates]", "float" ]
    }
  }
}

output {
  stdout { codec => rubydebug }
}

systemctl stop logstash
logstash -f logstash.conf

# 其他节点模拟访问
curl http://10.0.0.114/ -H "18.97.14.83: 157.166.226.25" -H "18.97.14.83:157.166.226.25"

3.3:效果展示

3.4:将信息输出到es中

cp logstash.conf ../conf.bak/logstash.bak-geoip-stdout
root@Logstash:/etc/logstash/conf.d# cat logstash.conf 
# Sample Logstash configuration for creating a simple
# Beats -> Logstash -> Elasticsearch pipeline.

input {
  beats {
    port => 5044
    codec => json
  }
}

filter {
  if [log_type] == "access" {
    geoip {
      source => "http_x_real_ip"
      target => "geoip"
      database => "/etc/logstash/GeoLite2-City.mmdb"
      add_field => [ "[geoip][coordinates]", "%{[geoip][longitude]}" ]
      add_field => [ "[geoip][coordinates]", "%{[geoip][latitude]}" ]
    }
    mutate {
      convert => [ "[geoip][coordinates]", "float" ]
    }
  }
}

output {
  if [log_type] == "access" {
    elasticsearch {
      hosts => ["10.0.0.110:9200"]
      index => "logstash-nginx-access-%{+yyy.MM.dd}"
    }
  }

  if [log_type] == "error" {
    elasticsearch {
      hosts => ["10.0.0.110:9200"]
      index => "logstash-nginx-error-%{+yyyy.MM.dd}"
    }
  }
}

systemctl restart logstash.service

3.5:效果展示

0

评论区