ELK相关笔记05-logstash的使用

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 "3.22.240.205: 157.166.226.25" -H "3.22.240.205: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：效果展示