Elastic Stack-04-Filebeat-Logstash-Elasticsearch

谈阎宝

2023-12-01

一、介绍

此篇文档的目的是把前几篇所说的知识点联系起来，做一个综合的使用，最终行称一个可以用于生产的简单配置方案。

使用 Filebeat 搜集 Nginx 的日志，输出到 Logstash, 再由 Logstash 处理完数据后输出到 Elasticsearch。

二、配置步骤

1 Elasticsearch

节点1 的配置文件 elasticsearch.yml内容

cluster.name: elk
node.name: ela1
node.data: true
network.host: 0.0.0.0
http.port: 9200
discovery.seed_hosts:
   - 192.168.122.6
   - 192.168.122.106:9300
   - 192.168.122.218
cluster.initial_master_nodes: ["ela1", "ela2", "ela3"]

节点2 的配置文件 elasticsearch.yml内容

cluster.name: elk
node.name: ela2
node.data: true
network.host: 0.0.0.0
http.port: 9200

discovery.seed_hosts:
   - 192.168.122.6
   - 192.168.122.106:9300
   - 192.168.122.218
cluster.initial_master_nodes: [ela1, ela2, ela3]

节点3 的配置文件 elasticsearch.yml内容

cluster.name: elk
node.name: ela3
node.data: true
network.host: 0.0.0.0
http.port: 9200

discovery.seed_hosts:
   - 192.168.122.6
   - 192.168.122.106:9300
   - 192.168.122.218
cluster.initial_master_nodes: [ela1, ela2, ela3]

2 Logstash

first-pipeline.conf

input {
   beats {
      port => 5044
   }
}

filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}" }
        remove_field => [ "message" ]
    }
    geoip {
       source => "clientip"
    }
}

output {
  stdout {
     codec => rubydebug
  }

  elasticsearch {
    # 这里是输出到 elasticsearch 集群中
    hosts => ["192.168.122.6:9200","192.168.122.106:9200","192.168.122.218:9200"]
  }
}

3 Filebeat

filebeat.yml

filebeat.inputs:
- type: log
  enabled: false
  paths:
    - /var/log/*.log
- type: filestream
  enabled: false
  paths:
    - /var/log/*.log
filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml
  reload.enabled: false
setup.template.settings:
  index.number_of_shards: 1
setup.kibana:
output.logstash:
  hosts: ["localhost:5044"]
processors:
  - add_host_metadata:
      when.not.contains.tags: forwarded
  - add_cloud_metadata: ~
  - add_docker_metadata: ~
  - add_kubernetes_metadata: ~

modules.d/nginx.yml

- module: nginx
  access:
    enabled: true
    var.paths: ["/var/log/nginx/access.log"]
  error:
    enabled: true

三、启动

1 启动并验证 Elasticsearch 集群

集群中的每个节点都执行如下命令：

./bin/elasticsearch  -d  -p /tmp/pid

在任意一个节点验证集群状态

集群健康状态

[root@ela1 ~]# curl -X GET "192.168.122.6:9200/_cat/health"
1609235083 09:44:43 elk green 3 3 8 4 0 0 0 0 - 100.0%

集群节点信息

[root@ela1 ~]# curl -X GET "192.168.122.6:9200/_cat/nodes"
192.168.122.6   20 83 0 0.11 0.10 0.07 cdhilmrstw - ela1
192.168.122.218 12 88 0 0.02 0.02 0.05 cdhilmrstw - ela3
192.168.122.106 45 92 0 0.00 0.01 0.05 cdhilmrstw * ela2

2 启动 Logstash

假如没有启动，或者没有自动加载配置文件，需要重新启动

./bin/logstash -f first-pipeline.conf --config.reload.automatic

3 启动 Filebeat

假如没有启动，需要重新启动

nohup   ./filebeat  &

四、验证

1 持续向日志文件中输入日志内容

执行如下命令，持续向你所需要收集的日志文件中输出内容。

注意修改最后的日志文件路径为正确的日志文件路径。

echo '10.9.29.211 - - [29/Dec/2020:17:43:09 +0800] "GET /35949/bundles/plugin/uptime/uptime.plugin.js HTTP/1.1" 200 5869 "http://10.9.12.250:8080/app/home" "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.116 Safari/537.36 QBCore/4.0.1301.400 QQBrowser/9.0.2524.400 Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/81.0.4044.138 Safari/537.36 NetType/WIFI MicroMessenger/7.0.20.1781(0x6700143B) WindowsWechat" "-"' >> /var/log/nginx/access.log

2 查看 Logstash 的终端

应用有相应的输出

{
        "fileset" => {
        "name" => "access"
    },
            "log" => {
        "offset" => 193560,
          "file" => {
            "path" => "/var/log/nginx/access.log"
        }
    },
           "host" => {
                  "mac" => [
            [ 0] "78:2b:cb:64:4e:25",
            [ 1] "78:2b:cb:64:4e:27",
            [ 2] "78:2b:cb:64:4e:29",
            [ 3] "78:2b:cb:64:4e:2b"
        ],
             "hostname" => "prod",
                 "name" => "prod",
        "containerized" => false,
                   "os" => {
            "codename" => "Core",
                "name" => "CentOS Linux",
              "family" => "redhat",
             "version" => "7 (Core)",
            "platform" => "centos",
              "kernel" => "3.10.0-957.el7.x86_64"
        },
                   "ip" => [
            [ 0] "10.9.12.250",
            [ 1] "fe80::7a2b:cbff:fe64:4e25",
            [ 2] "192.168.122.1"
        ],
                   "id" => "e14bb5e04254411da829c0bc6beedbf3",
         "architecture" => "x86_64"
    },
        "request" => "/35949/bundles/plugin/uptime/uptime.plugin.js",
           "auth" => "-",
            "ecs" => {
        "version" => "1.5.0"
    },
     "@timestamp" => 2020-12-29T09:46:49.093Z,
          "agent" => {
            "hostname" => "prod",
                "name" => "prod",
             "version" => "7.10.0",
                "type" => "filebeat",
        "ephemeral_id" => "bfeb1d8f-d791-4706-9001-8ea26bd7557e",
                  "id" => "153bb637-ba53-4b3e-888d-7f60db4aa877"
    },
        "service" => {
        "type" => "nginx"
    },
           "verb" => "GET",
          "bytes" => "5869",
       "referrer" => "\"http://10.9.12.250:8080/app/home\"",
          "geoip" => {},
          "ident" => "-",
       "@version" => "1",
      "timestamp" => "29/Dec/2020:17:43:09 +0800",
       "response" => "200",
           "tags" => [
        [0] "beats_input_codec_plain_applied",
        [1] "_geoip_lookup_failure"
    ],
          "input" => {
        "type" => "log"
    },
       "clientip" => "10.9.29.211",
    "httpversion" => "1.1",
          "event" => {
         "dataset" => "nginx.access",
          "module" => "nginx",
        "timezone" => "+08:00"
    }
}

3 验证Elasticsearch是否创建了索引

[root@prod logstash]# curl -X GET "192.168.122.6:9200/_cat/indices"
green open logstash-2020.12.29-000001 IgYwIwXVTeSqc6f4SGAa3Q 1 1 2 0 118.4kb 59.2kb

目前 logstash-2020.12.29-000001 是Elasticsearch 自动创建的索引。

五、创建自己需要的索引

从之前的查询中知道，索引是 Elasticsearch 是自动创建的，但是，这往往不是生产中想要的。

比较好的一个建议是：

把正确的访问日志 access.log 中的内容单独放到一个索引中。
把错误日志 error.log 中的内容单独放到另外一个索引中。

1 更新配置文件

未使用 Filebeat 内置模块的情况

input {
   beats {
      port => 5044
   }
}

filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}" }
        remove_field => [ "message" ]
    }
    geoip {
       source => "clientip"
    }
}
output {
  stdout {
     codec => rubydebug
  }

  if [log][file][path] == "/tmp/access.log" {
    elasticsearch {
      hosts => ["192.168.122.6:9200","192.168.122.106:9200","192.168.122.25:9200"]
      index => "%{[host][hostname]}-nginx-access-%{+YYYY.MM.dd}"
    }
  } else if [log][file][path] == "/tmp/error.log" {
    elasticsearch {
      hosts => ["192.168.122.6:9200","192.168.122.106:9200","192.168.122.25:9200"]
      index => "%{[host][hostname]}-nginx-error-%{+YYYY.MM.dd}"
    }
  }
}

使用 Filebeat 内置模块的情况

input {
   beats {
      port => 5044
   }
}

filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}" }
        remove_field => [ "message" ]
    }
    geoip {
       source => "clientip"
    }
}

output {
  stdout {
     codec => rubydebug
  }
  if [event][dataset] == "nginx.access" {
    elasticsearch {
      # 这里是输出到 elasticsearch 集群中
      hosts => ["192.168.122.6:9200","192.168.122.106:9200","192.168.122.218:9200"]
      # 创建索引，索引好比是MySQL 数据库中的表名称
      index => "%{[host][hostname]}-nginx-access-%{+YYYY.MM.dd}"
    }
  } else if  [event][dataset] == "nginx.error" {
  elasticsearch {
      # 这里是输出到 elasticsearch 集群中
      hosts => ["192.168.122.6:9200","192.168.122.106:9200","192.168.122.218:9200"]

      # 创建索引，索引好比是MySQL 数据库中的表名称
      index => "%{[host][hostname]}-nginx-error-%{+YYYY.MM.dd}"
    }
  }
}

2 添加新的日志内容到相关日志文件

分别向两个测试的日志文件中添加新的日志内容

向 /tmp/asscess.log 添加内容


echo '10.9.29.211 - - [29/Dec/2020:17:43:09 +0800] "GET /35949/bundles/plugin/uptime/uptime.plugin.js HTTP/1.1" 200 5869 "http://10.9.12.250:8080/app/home" "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.116 Safari/537.36 QBCore/4.0.1301.400 QQBrowser/9.0.2524.400 Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/81.0.4044.138 Safari/537.36 NetType/WIFI MicroMessenger/7.0.20.1781(0x6700143B) WindowsWechat" "-"' >> /var/log/nginx/access.log

向 /tmp/error.log 添加内容

echo '2020/12/29 15:25:10 [warn] 3380#3380: *161 an upstream response is buffered to a temporary file /var/cache/nginx/proxy_temp/5/00/0000000005 while reading upstream, client: 10.9.29.234, server: localhost, request: "GET /35949/bundles/plugin/data/data.plugin.js HTTP/1.1", upstream: "http://127.0.0.1:5601/35949/bundles/plugin/data/data.plugin.js", host: "10.9.12.250:8080", referrer: "http://10.9.12.250:8080/app/home"' >> /var/log/nginx/error.log

3 验证 Elasticsearch 集群中的索引

[root@ela1 ~]# curl -X GET "192.168.122.6:9200/_cat/indices"
green open logstash-ela1-2020.12.29-000001 IgYwIwXVTeSqc6f4SGAa3Q 1 1 212 0  1.1mb 648.6kb
green open ela1-nginx-access-2020.12.29    ien9RLwPTM2w0P_Vebrr3w 1 1   1 0 66.6kb  33.3kb
green open ela1-nginx-error-2020.12.29     Bn0SV7i7STGVANn4x0mj4g 1 1   1 0 56.2kb  28.1kb

logstash-2020.12.29-000001 是原来的

ela1-nginx-access-2020.12.29 和 ela1-nginx-error-2020.12.29 是新建立的。

Elastic Stack-04-Filebeat-Logstash-Elasticsearch

一、介绍

二、配置步骤

1 Elasticsearch

2 Logstash

3 Filebeat

三、启动

1 启动并验证 Elasticsearch 集群

2 启动 Logstash

3 启动 Filebeat

四、验证

1 持续向日志文件中输入日志内容

2 查看 Logstash 的终端

3 验证Elasticsearch是否创建了索引

五、创建自己需要的索引

1 更新配置文件

未使用 Filebeat 内置模块的情况

使用 Filebeat 内置模块的情况

2 添加新的日志内容到相关日志文件

3 验证 Elasticsearch 集群中的索引

相关阅读

相关文章

相关问答

相关文档

Elastic Stack-04-Filebeat-Logstash-Elasticsearch

一、 介绍

二、配置步骤

1 Elasticsearch

2 Logstash

3 Filebeat

三、启动

1 启动并验证 Elasticsearch 集群

2 启动 Logstash

3 启动 Filebeat

四、 验证

1 持续向日志文件中输入日志内容

2 查看 Logstash 的终端

3 验证Elasticsearch是否创建了索引

五、创建自己需要的索引

1 更新配置文件

未使用 Filebeat 内置模块的情况

使用 Filebeat 内置模块的情况

2 添加新的日志内容到相关日志文件

3 验证 Elasticsearch 集群中的索引

相关阅读

相关文章

相关问答

相关文档

一、介绍

四、验证